From 3e47e45230596a3fc8a6937109ea90680751115c Mon Sep 17 00:00:00 2001 From: zouxxyy Date: Wed, 2 Sep 2026 21:38:50 +0800 Subject: [PATCH 1/6] [spark] Support reading and writing Fluss tables through Paimon catalog Route Paimon tables marked with lakestream.enabled to the Fluss Spark catalog and cover log and primary-key table reads and writes on Spark 3.4 and 3.5. Co-Authored-By: Claude Code AI-Model: gpt-5 Co-Authored-By: Codex AI-Contributed/Feature: 329/329 AI-Contributed/UT: 471/471 --- paimon-spark/paimon-spark-3.4/pom.xml | 57 +++++ .../spark/FlussCatalogReadWriteITCase.scala | 21 ++ .../paimon/spark/FlussCatalogTest.scala | 21 ++ paimon-spark/paimon-spark-3.5/pom.xml | 57 +++++ .../spark/FlussCatalogReadWriteITCase.scala | 21 ++ .../paimon/spark/FlussCatalogTest.scala | 21 ++ .../paimon/spark/FlussCatalogDelegate.java | 126 +++++++++++ .../org/apache/paimon/spark/SparkCatalog.java | 8 +- .../paimon/spark/catalog/SupportFluss.java | 42 ++++ paimon-spark/paimon-spark-ut/pom.xml | 37 ++++ .../FlussCatalogReadWriteITCaseBase.scala | 182 ++++++++++++++++ .../paimon/spark/FlussCatalogTestBase.scala | 205 ++++++++++++++++++ pom.xml | 2 + 13 files changed, 799 insertions(+), 1 deletion(-) create mode 100644 paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/FlussCatalogReadWriteITCase.scala create mode 100644 paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/FlussCatalogTest.scala create mode 100644 paimon-spark/paimon-spark-3.5/src/test/scala/org/apache/paimon/spark/FlussCatalogReadWriteITCase.scala create mode 100644 paimon-spark/paimon-spark-3.5/src/test/scala/org/apache/paimon/spark/FlussCatalogTest.scala create mode 100644 paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/FlussCatalogDelegate.java create mode 100644 paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/SupportFluss.java create mode 100644 paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/FlussCatalogReadWriteITCaseBase.scala create mode 100644 paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/FlussCatalogTestBase.scala diff --git a/paimon-spark/paimon-spark-3.4/pom.xml b/paimon-spark/paimon-spark-3.4/pom.xml index ffa65a71dba2..827d8e763a33 100644 --- a/paimon-spark/paimon-spark-3.4/pom.xml +++ b/paimon-spark/paimon-spark-3.4/pom.xml @@ -110,6 +110,63 @@ under the License. tests test + + + org.apache.fluss + fluss-spark-3.4_${scala.binary.version} + ${fluss.version} + test + + + + org.apache.fluss + fluss-server + ${fluss.version} + test + + + + org.apache.fluss + fluss-server + ${fluss.version} + test-jar + test + + + + org.apache.fluss + fluss-common + ${fluss.version} + test-jar + test + + + + org.apache.fluss + fluss-test-utils + ${fluss.version} + test + + + + org.apache.fluss + fluss-lake-paimon + ${fluss.version} + test + + + org.apache.paimon + * + + + + + + org.apache.curator + curator-test + ${fluss.curator.version} + test + diff --git a/paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/FlussCatalogReadWriteITCase.scala b/paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/FlussCatalogReadWriteITCase.scala new file mode 100644 index 000000000000..c03e1b32d59e --- /dev/null +++ b/paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/FlussCatalogReadWriteITCase.scala @@ -0,0 +1,21 @@ +/* + * 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 + +class FlussCatalogReadWriteITCase extends FlussCatalogReadWriteITCaseBase {} diff --git a/paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/FlussCatalogTest.scala b/paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/FlussCatalogTest.scala new file mode 100644 index 000000000000..ec8e47a8af4d --- /dev/null +++ b/paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/FlussCatalogTest.scala @@ -0,0 +1,21 @@ +/* + * 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 + +class FlussCatalogTest extends FlussCatalogTestBase {} diff --git a/paimon-spark/paimon-spark-3.5/pom.xml b/paimon-spark/paimon-spark-3.5/pom.xml index 6d72e2fbaa08..43f838ff84bf 100644 --- a/paimon-spark/paimon-spark-3.5/pom.xml +++ b/paimon-spark/paimon-spark-3.5/pom.xml @@ -110,6 +110,63 @@ under the License. tests test + + + org.apache.fluss + fluss-spark-3.5_${scala.binary.version} + ${fluss.version} + test + + + + org.apache.fluss + fluss-server + ${fluss.version} + test + + + + org.apache.fluss + fluss-server + ${fluss.version} + test-jar + test + + + + org.apache.fluss + fluss-common + ${fluss.version} + test-jar + test + + + + org.apache.fluss + fluss-test-utils + ${fluss.version} + test + + + + org.apache.fluss + fluss-lake-paimon + ${fluss.version} + test + + + org.apache.paimon + * + + + + + + org.apache.curator + curator-test + ${fluss.curator.version} + test + diff --git a/paimon-spark/paimon-spark-3.5/src/test/scala/org/apache/paimon/spark/FlussCatalogReadWriteITCase.scala b/paimon-spark/paimon-spark-3.5/src/test/scala/org/apache/paimon/spark/FlussCatalogReadWriteITCase.scala new file mode 100644 index 000000000000..c03e1b32d59e --- /dev/null +++ b/paimon-spark/paimon-spark-3.5/src/test/scala/org/apache/paimon/spark/FlussCatalogReadWriteITCase.scala @@ -0,0 +1,21 @@ +/* + * 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 + +class FlussCatalogReadWriteITCase extends FlussCatalogReadWriteITCaseBase {} diff --git a/paimon-spark/paimon-spark-3.5/src/test/scala/org/apache/paimon/spark/FlussCatalogTest.scala b/paimon-spark/paimon-spark-3.5/src/test/scala/org/apache/paimon/spark/FlussCatalogTest.scala new file mode 100644 index 000000000000..ec8e47a8af4d --- /dev/null +++ b/paimon-spark/paimon-spark-3.5/src/test/scala/org/apache/paimon/spark/FlussCatalogTest.scala @@ -0,0 +1,21 @@ +/* + * 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 + +class FlussCatalogTest extends FlussCatalogTestBase {} diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/FlussCatalogDelegate.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/FlussCatalogDelegate.java new file mode 100644 index 000000000000..27ff9d2a982b --- /dev/null +++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/FlussCatalogDelegate.java @@ -0,0 +1,126 @@ +/* + * 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; + +import org.apache.spark.sql.catalyst.analysis.NoSuchTableException; +import org.apache.spark.sql.connector.catalog.Identifier; +import org.apache.spark.sql.connector.catalog.Table; +import org.apache.spark.sql.connector.catalog.TableCatalog; +import org.apache.spark.sql.util.CaseInsensitiveStringMap; + +import java.util.HashMap; +import java.util.Map; +import java.util.stream.Collectors; + +/** Loads LakeStream tables from the Fluss Spark catalog used by Spark reads and writes. */ +class FlussCatalogDelegate { + + static final String FLUSS_OPTION_PREFIX = "fluss."; + static final String FLUSS_BOOTSTRAP_SERVERS = "fluss.bootstrap.servers"; + + private static final String FLUSS_CATALOG_CLASS = "org.apache.fluss.spark.SparkCatalog"; + + private final String catalogName; + private final Map flussOptions; + private final CatalogLoader catalogLoader; + + private volatile TableCatalog flussCatalog; + + FlussCatalogDelegate(Map catalogOptions, String catalogName) { + this(catalogOptions, catalogName, FlussCatalogDelegate::loadFlussCatalog); + } + + FlussCatalogDelegate( + Map catalogOptions, String catalogName, CatalogLoader catalogLoader) { + this.catalogName = catalogName; + this.flussOptions = + catalogOptions.entrySet().stream() + .filter(entry -> entry.getKey().startsWith(FLUSS_OPTION_PREFIX)) + .collect( + Collectors.toMap( + entry -> + entry.getKey() + .substring(FLUSS_OPTION_PREFIX.length()), + Map.Entry::getValue)); + this.catalogLoader = catalogLoader; + } + + boolean isConfigured() { + return flussOptions.containsKey("bootstrap.servers"); + } + + Map flussOptions() { + return new HashMap<>(flussOptions); + } + + Table loadTable(Identifier identifier) throws NoSuchTableException { + return catalog().loadTable(identifier); + } + + private TableCatalog catalog() { + if (flussCatalog == null) { + synchronized (this) { + if (flussCatalog == null) { + if (!isConfigured()) { + throw new IllegalStateException( + String.format( + "Paimon catalog '%s' contains a Fluss LakeStream table, " + + "but Fluss access is not configured. Add '%s' " + + "and any required 'fluss.client.security.*' " + + "options to the Paimon catalog.", + catalogName, FLUSS_BOOTSTRAP_SERVERS)); + } + try { + TableCatalog catalog = catalogLoader.load(contextClassLoader()); + catalog.initialize(catalogName, new CaseInsensitiveStringMap(flussOptions)); + flussCatalog = catalog; + } catch (Exception e) { + throw new IllegalStateException( + String.format( + "Failed to create the Fluss delegate for Paimon catalog " + + "'%s'. Make sure a Fluss Spark connector " + + "matching the Spark version is on the classpath.", + catalogName), + e); + } + } + } + } + return flussCatalog; + } + + private static TableCatalog loadFlussCatalog(ClassLoader classLoader) throws Exception { + Class catalogClass = Class.forName(FLUSS_CATALOG_CLASS, true, classLoader); + Object catalog = catalogClass.getDeclaredConstructor().newInstance(); + if (!(catalog instanceof TableCatalog)) { + throw new IllegalStateException( + FLUSS_CATALOG_CLASS + " does not implement Spark TableCatalog."); + } + return (TableCatalog) catalog; + } + + private static ClassLoader contextClassLoader() { + ClassLoader classLoader = Thread.currentThread().getContextClassLoader(); + return classLoader == null ? FlussCatalogDelegate.class.getClassLoader() : classLoader; + } + + interface CatalogLoader { + TableCatalog load(ClassLoader classLoader) throws Exception; + } +} diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java index 6fe7ea50335e..ee224dbaf30a 100644 --- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java +++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java @@ -32,6 +32,7 @@ import org.apache.paimon.schema.SchemaChange; import org.apache.paimon.spark.catalog.FormatTableCatalog; import org.apache.paimon.spark.catalog.SparkBaseCatalog; +import org.apache.paimon.spark.catalog.SupportFluss; import org.apache.paimon.spark.catalog.SupportV1Function; import org.apache.paimon.spark.catalog.SupportView; import org.apache.paimon.spark.catalog.functions.FunctionIdentifierConverter; @@ -117,6 +118,7 @@ /** Spark {@link TableCatalog} for paimon. */ public class SparkCatalog extends SparkBaseCatalog implements SupportView, + SupportFluss, SupportV1Function, FunctionCatalog, SupportsNamespaces, @@ -130,6 +132,7 @@ public class SparkCatalog extends SparkBaseCatalog private Catalog catalog; private String defaultDatabase; private boolean v1FunctionEnabled; + private FlussCatalogDelegate flussCatalogDelegate; @Nullable private PaimonV1FunctionRegistry v1FunctionRegistry; @Override @@ -137,6 +140,7 @@ public void initialize(String name, CaseInsensitiveStringMap options) { SparkSession sparkSession = PaimonSparkSession$.MODULE$.active(); checkRequiredConfigurations(sparkSession); this.catalogName = name; + this.flussCatalogDelegate = new FlussCatalogDelegate(options.asCaseSensitiveMap(), name); CatalogContext catalogContext = CatalogContext.create( Options.fromMap(options.asCaseSensitiveMap()), @@ -309,7 +313,9 @@ public void invalidateTable(Identifier ident) { @Override public org.apache.spark.sql.connector.catalog.Table loadTable(Identifier ident) throws NoSuchTableException { - return loadSparkTable(ident, Collections.emptyMap()); + org.apache.spark.sql.connector.catalog.Table table = + loadSparkTable(ident, Collections.emptyMap()); + return isFlussTable(table) ? flussCatalogDelegate.loadTable(ident) : table; } /** diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/SupportFluss.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/SupportFluss.java new file mode 100644 index 000000000000..7d79227454ab --- /dev/null +++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/SupportFluss.java @@ -0,0 +1,42 @@ +/* + * 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.catalog; + +import org.apache.paimon.spark.SparkTable; +import org.apache.paimon.table.FileStoreTable; + +import org.apache.spark.sql.connector.catalog.Table; + +/** Catalog support for Fluss LakeStream tables backed by Paimon. */ +public interface SupportFluss { + + String LAKESTREAM_ENABLED = "lakestream.enabled"; + + default boolean isFlussTable(Table table) { + if (!(table instanceof SparkTable)) { + return false; + } + return isFlussTable(((SparkTable) table).getTable()); + } + + default boolean isFlussTable(org.apache.paimon.table.Table table) { + return table instanceof FileStoreTable + && Boolean.parseBoolean(table.options().get(LAKESTREAM_ENABLED)); + } +} diff --git a/paimon-spark/paimon-spark-ut/pom.xml b/paimon-spark/paimon-spark-ut/pom.xml index ee8564dd66bc..8f616bb8b199 100644 --- a/paimon-spark/paimon-spark-ut/pom.xml +++ b/paimon-spark/paimon-spark-ut/pom.xml @@ -102,6 +102,43 @@ under the License. ${mockito.version} test + + + org.apache.fluss + fluss-server + ${fluss.version} + test + + + + org.apache.fluss + fluss-server + ${fluss.version} + test-jar + test + + + + org.apache.fluss + fluss-common + ${fluss.version} + test-jar + test + + + + org.apache.fluss + fluss-test-utils + ${fluss.version} + test + + + + org.apache.curator + curator-test + ${fluss.curator.version} + test + diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/FlussCatalogReadWriteITCaseBase.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/FlussCatalogReadWriteITCaseBase.scala new file mode 100644 index 000000000000..19bbcbac945f --- /dev/null +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/FlussCatalogReadWriteITCaseBase.scala @@ -0,0 +1,182 @@ +/* + * 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 + +import org.apache.paimon.catalog.Identifier +import org.apache.paimon.schema.SchemaChange +import org.apache.paimon.spark.catalog.SupportFluss +import org.apache.paimon.utils.FileIOUtils + +import org.apache.fluss.config.{ConfigOptions, Configuration} +import org.apache.fluss.server.testutils.FlussClusterExtension +import org.apache.spark.SparkConf +import org.apache.spark.sql.Row + +import java.nio.file.{Files, Path} +import java.time.Duration + +/** MiniCluster tests for reading and writing Fluss LakeStream tables through Paimon catalog. */ +abstract class FlussCatalogReadWriteITCaseBase extends PaimonSparkTestBase { + + private val flussCatalogName = "fluss_catalog" + private val flussDatabase = "fluss" + private val lakeWarehouse: Path = + Files.createTempDirectory("paimon-fluss-catalog-it").resolve("warehouse") + + private val flussCluster: FlussClusterExtension = + FlussClusterExtension.builder + .setClusterConf(flussConfiguration) + .setNumOfTabletServers(1) + .build + + override protected val dbName0: String = flussDatabase + + override protected def sparkConf: SparkConf = { + val conf = super.sparkConf + val bootstrapServers = flussCluster.getBootstrapServers + val paimonExtensions = conf.get("spark.sql.extensions") + + conf + .set(s"spark.sql.catalog.$flussCatalogName", "org.apache.fluss.spark.SparkCatalog") + .set(s"spark.sql.catalog.$flussCatalogName.bootstrap.servers", bootstrapServers) + .set(s"spark.sql.catalog.$flussCatalogName.datalake.format", "paimon") + .set(s"spark.sql.catalog.$flussCatalogName.datalake.paimon.metastore", "filesystem") + .set(s"spark.sql.catalog.$flussCatalogName.datalake.paimon.warehouse", lakeWarehouse.toString) + .set("spark.sql.catalog.paimon.warehouse", lakeWarehouse.toString) + .set("spark.sql.catalog.paimon.fluss.bootstrap.servers", bootstrapServers) + .set("spark.sql.catalog.paimon.fluss.datalake.format", "paimon") + .set("spark.sql.catalog.paimon.fluss.datalake.paimon.metastore", "filesystem") + .set("spark.sql.catalog.paimon.fluss.datalake.paimon.warehouse", lakeWarehouse.toString) + .set( + "spark.sql.extensions", + s"$paimonExtensions,org.apache.fluss.spark.FlussSparkSessionExtensions") + } + + override protected def beforeAll(): Unit = { + flussCluster.start() + try { + super.beforeAll() + } catch { + case t: Throwable => + flussCluster.close() + throw t + } + } + + override protected def afterAll(): Unit = { + try { + super.afterAll() + } finally { + try { + flussCluster.close() + } finally { + FileIOUtils.deleteDirectoryQuietly(lakeWarehouse.getParent.toFile) + } + } + } + + test("read and write a Fluss log table through Paimon catalog") { + withFlussTable("log_orders") { + verifyLakeStreamMarker("log_orders") + + sql("INSERT INTO paimon.fluss.log_orders VALUES (1, 'a'), (2, 'b')") + checkAnswer( + sql("SELECT * FROM fluss_catalog.fluss.log_orders"), + Row(1, "a") :: Row(2, "b") :: Nil) + + sql("INSERT INTO fluss_catalog.fluss.log_orders VALUES (3, 'c')") + checkAnswer( + sql("SELECT * FROM paimon.fluss.log_orders"), + Row(1, "a") :: Row(2, "b") :: Row(3, "c") :: Nil) + } + } + + test("read and write a Fluss primary-key table through Paimon catalog") { + withFlussTable("pk_orders", primaryKey = true) { + verifyLakeStreamMarker("pk_orders") + + sql("INSERT INTO paimon.fluss.pk_orders VALUES (1, 'a'), (2, 'b')") + checkAnswer( + sql("SELECT * FROM fluss_catalog.fluss.pk_orders"), + Row(1, "a") :: Row(2, "b") :: Nil) + + sql("INSERT INTO fluss_catalog.fluss.pk_orders VALUES (1, 'a2'), (3, 'c')") + checkAnswer( + sql("SELECT * FROM paimon.fluss.pk_orders"), + Row(1, "a2") :: Row(2, "b") :: Row(3, "c") :: Nil) + } + } + + private def withFlussTable(tableName: String, primaryKey: Boolean = false)(f: => Unit): Unit = { + val primaryKeyProperty = if (primaryKey) ", 'primary.key' = 'id'" else "" + val paimonPrimaryKeyProperties = + if (primaryKey) { + ", 'primary-key' = 'id', 'bucket' = '1', 'bucket-key' = 'id', " + + "'changelog-producer' = 'input'" + } else { + ", 'bucket' = '-1'" + } + // Model the externally managed lifecycle: materialize matching Paimon lake metadata first and + // publish the LakeStream marker only after the Fluss table exists. + sql(s""" + |CREATE TABLE paimon.$flussDatabase.$tableName (id INT, name STRING) + |TBLPROPERTIES ( + | 'partition.legacy-name' = 'false'$paimonPrimaryKeyProperties) + |""".stripMargin) + try { + sql(s""" + |CREATE TABLE $flussCatalogName.$flussDatabase.$tableName (id INT, name STRING) + |TBLPROPERTIES ( + | 'bucket.num' = '1', + | 'table.datalake.enabled' = 'true', + | 'table.datalake.format' = 'paimon', + | 'table.datalake.freshness' = '60s'$primaryKeyProperty) + |""".stripMargin) + paimonCatalog.alterTable( + Identifier.create(flussDatabase, tableName), + java.util.Collections.singletonList( + SchemaChange.setOption(SupportFluss.LAKESTREAM_ENABLED, "true")), + false + ) + try { + f + } finally { + sql(s"DROP TABLE IF EXISTS $flussCatalogName.$flussDatabase.$tableName") + } + } finally { + sql(s"DROP TABLE IF EXISTS paimon.$flussDatabase.$tableName") + } + } + + private def verifyLakeStreamMarker(tableName: String): Unit = { + val table = loadTable(flussDatabase, tableName) + assert(table.options().get(SupportFluss.LAKESTREAM_ENABLED) == "true") + } + + private def flussConfiguration: Configuration = { + val conf = new Configuration + conf.set(ConfigOptions.KV_SNAPSHOT_INTERVAL, Duration.ofSeconds(1)) + conf.setString("datalake.format", "paimon") + conf.setString("datalake.paimon.metastore", "filesystem") + conf.setString("datalake.paimon.cache-enabled", "false") + conf.setString("datalake.paimon.warehouse", lakeWarehouse.toString) + conf.setString("server.data-disk.write-limit-ratio", "1.0") + conf + } +} diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/FlussCatalogTestBase.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/FlussCatalogTestBase.scala new file mode 100644 index 000000000000..742df3a0a61f --- /dev/null +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/FlussCatalogTestBase.scala @@ -0,0 +1,205 @@ +/* + * 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 + +import org.apache.paimon.spark.catalog.SupportFluss +import org.apache.paimon.table.FileStoreTable + +import org.apache.spark.sql.connector.catalog.{Identifier, Table, TableCapability, TableCatalog, TableChange} +import org.apache.spark.sql.connector.expressions.Transform +import org.apache.spark.sql.types.StructType +import org.apache.spark.sql.util.CaseInsensitiveStringMap + +import java.lang.reflect.{Field, InvocationHandler, Method, Proxy} +import java.util.{Collections, HashMap, Map => JMap, Set => JSet} +import java.util.concurrent.atomic.AtomicReference + +import scala.collection.JavaConverters._ + +/** Tests Fluss LakeStream delegation through the Paimon Spark catalog. */ +abstract class FlussCatalogTestBase extends PaimonSparkTestBase { + + private val tableIdentifier = Identifier.of(Array("db"), "orders") + + test("map Fluss options and initialize the delegate lazily") { + val catalog = new TestingTableCatalog + val loadedClassLoader = new AtomicReference[ClassLoader] + val options = new HashMap[String, String] + options.put("warehouse", "/tmp/warehouse") + options.put("fluss.bootstrap.servers", "localhost:9123") + options.put("fluss.client.security.protocol", "sasl") + val delegate = new FlussCatalogDelegate( + options, + "paimon", + new FlussCatalogDelegate.CatalogLoader { + override def load(classLoader: ClassLoader): TableCatalog = { + loadedClassLoader.set(classLoader) + catalog + } + } + ) + + assert( + delegate.flussOptions().asScala == Map( + "bootstrap.servers" -> "localhost:9123", + "client.security.protocol" -> "sasl")) + assert(delegate.loadTable(tableIdentifier) eq catalog.table) + assert(catalog.catalogName == "paimon") + assert( + catalog.options.asCaseSensitiveMap().asScala == Map( + "bootstrap.servers" -> "localhost:9123", + "client.security.protocol" -> "sasl")) + assert(loadedClassLoader.get() != null) + + delegate.loadTable(tableIdentifier) + assert(catalog.initializeCount == 1) + } + + test("require Fluss bootstrap servers") { + val delegate = new FlussCatalogDelegate(Collections.emptyMap[String, String](), "paimon") + + val error = intercept[IllegalStateException] { + delegate.loadTable(tableIdentifier) + } + assert(error.getMessage.contains("fluss.bootstrap.servers")) + } + + test("route a marked file store table to Fluss") { + val paimonTable = + SparkTable(fileStoreTable(Collections.singletonMap(SupportFluss.LAKESTREAM_ENABLED, "true"))) + val flussCatalog = new TestingTableCatalog + val catalog = testingSparkCatalog(paimonTable, configuredDelegate(flussCatalog)) + + assert(catalog.loadTable(tableIdentifier) eq flussCatalog.table) + assert(flussCatalog.initializeCount == 1) + } + + test("keep a regular file store table on Paimon") { + val paimonTable = SparkTable(fileStoreTable(Collections.emptyMap[String, String]())) + val flussCatalog = new TestingTableCatalog + val catalog = testingSparkCatalog(paimonTable, configuredDelegate(flussCatalog)) + + assert(catalog.loadTable(tableIdentifier) eq paimonTable) + assert(flussCatalog.initializeCount == 0) + } + + test("do not route a non-file-store table carrying the marker") { + val systemTable = + new TestingTable(Collections.singletonMap(SupportFluss.LAKESTREAM_ENABLED, "true")) + val flussCatalog = new TestingTableCatalog + val catalog = testingSparkCatalog(systemTable, configuredDelegate(flussCatalog)) + + assert(catalog.loadTable(tableIdentifier) eq systemTable) + assert(flussCatalog.initializeCount == 0) + } + + private def configuredDelegate(catalog: TableCatalog): FlussCatalogDelegate = + new FlussCatalogDelegate( + Collections.singletonMap("fluss.bootstrap.servers", "localhost:9123"), + "paimon", + new FlussCatalogDelegate.CatalogLoader { + override def load(classLoader: ClassLoader): TableCatalog = catalog + } + ) + + private def testingSparkCatalog(table: Table, delegate: FlussCatalogDelegate): SparkCatalog = { + val catalog = new TestingSparkCatalog(table) + val field: Field = classOf[SparkCatalog].getDeclaredField("flussCatalogDelegate") + field.setAccessible(true) + field.set(catalog, delegate) + catalog + } + + private def fileStoreTable(options: JMap[String, String]): FileStoreTable = + Proxy + .newProxyInstance( + classOf[FileStoreTable].getClassLoader, + Array(classOf[FileStoreTable]), + new InvocationHandler { + override def invoke(proxy: Object, method: Method, args: Array[Object]): Object = + method.getName match { + case "options" => options + case "toString" => "TestingFileStoreTable" + case _ => throw new UnsupportedOperationException(method.toString) + } + } + ) + .asInstanceOf[FileStoreTable] + + private class TestingSparkCatalog(table: Table) extends SparkCatalog { + + override protected def loadSparkTable( + ident: Identifier, + extraOptions: JMap[String, String]): Table = table + } + + private class TestingTable(tableProperties: JMap[String, String]) extends Table { + + override def name(): String = "orders" + + override def schema(): StructType = new StructType + + override def capabilities(): JSet[TableCapability] = + Collections.emptySet[TableCapability]() + + override def properties(): JMap[String, String] = tableProperties + } + + private class TestingTableCatalog extends TableCatalog { + + val table: Table = new TestingTable(Collections.emptyMap[String, String]()) + + var catalogName: String = _ + var options: CaseInsensitiveStringMap = _ + var initializeCount: Int = 0 + var lastIdentifier: Identifier = _ + + override def initialize(name: String, options: CaseInsensitiveStringMap): Unit = { + catalogName = name + this.options = options + initializeCount += 1 + } + + override def name(): String = catalogName + + override def listTables(namespace: Array[String]): Array[Identifier] = + Array(tableIdentifier) + + override def loadTable(identifier: Identifier): Table = { + lastIdentifier = identifier + table + } + + override def createTable( + identifier: Identifier, + schema: StructType, + partitions: Array[Transform], + properties: JMap[String, String]): Table = + throw new UnsupportedOperationException + + override def alterTable(identifier: Identifier, changes: TableChange*): Table = + throw new UnsupportedOperationException + + override def dropTable(identifier: Identifier): Boolean = + throw new UnsupportedOperationException + + override def renameTable(oldIdentifier: Identifier, newIdentifier: Identifier): Unit = + throw new UnsupportedOperationException + } +} diff --git a/pom.xml b/pom.xml index 5a6bf1a91640..c41481e66195 100644 --- a/pom.xml +++ b/pom.xml @@ -113,6 +113,8 @@ under the License. true 1.19.1 1.6.1 + 1.0-SNAPSHOT + 5.4.0 1.16.0 1.9.8 3.19.6 From 8073aa936cf8914cd2272a960337e8ab257105d8 Mon Sep 17 00:00:00 2001 From: zouxxyy Date: Wed, 2 Sep 2026 22:48:05 +0800 Subject: [PATCH 2/6] [spark] Limit Fluss connector tests to Scala 2.12 Use the published Fluss Spark 2.12 connector test artifacts for all builds because Scala 2.13 snapshots are unavailable. Skip MiniCluster integration tests under Scala 2.13 while keeping catalog routing tests enabled. Co-Authored-By: Claude Code AI-Model: gpt-5 Co-Authored-By: Codex AI-Contributed/Feature: 5/5 AI-Contributed/UT: 57/57 --- paimon-spark/paimon-spark-3.4/pom.xml | 2 +- paimon-spark/paimon-spark-3.5/pom.xml | 2 +- .../FlussCatalogReadWriteITCaseBase.scala | 57 ++++++++++++------- pom.xml | 1 + 4 files changed, 40 insertions(+), 22 deletions(-) diff --git a/paimon-spark/paimon-spark-3.4/pom.xml b/paimon-spark/paimon-spark-3.4/pom.xml index 827d8e763a33..4eb4b66080dd 100644 --- a/paimon-spark/paimon-spark-3.4/pom.xml +++ b/paimon-spark/paimon-spark-3.4/pom.xml @@ -113,7 +113,7 @@ under the License. org.apache.fluss - fluss-spark-3.4_${scala.binary.version} + fluss-spark-3.4_${fluss.scala.binary.version} ${fluss.version} test diff --git a/paimon-spark/paimon-spark-3.5/pom.xml b/paimon-spark/paimon-spark-3.5/pom.xml index 43f838ff84bf..32492ac1dbf0 100644 --- a/paimon-spark/paimon-spark-3.5/pom.xml +++ b/paimon-spark/paimon-spark-3.5/pom.xml @@ -113,7 +113,7 @@ under the License. org.apache.fluss - fluss-spark-3.5_${scala.binary.version} + fluss-spark-3.5_${fluss.scala.binary.version} ${fluss.version} test diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/FlussCatalogReadWriteITCaseBase.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/FlussCatalogReadWriteITCaseBase.scala index 19bbcbac945f..a75ea2b943aa 100644 --- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/FlussCatalogReadWriteITCaseBase.scala +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/FlussCatalogReadWriteITCaseBase.scala @@ -34,6 +34,9 @@ import java.time.Duration /** MiniCluster tests for reading and writing Fluss LakeStream tables through Paimon catalog. */ abstract class FlussCatalogReadWriteITCaseBase extends PaimonSparkTestBase { + private val flussSparkConnectorAvailable = + scala.util.Properties.versionNumberString.startsWith("2.12.") + private val flussCatalogName = "fluss_catalog" private val flussDatabase = "fluss" private val lakeWarehouse: Path = @@ -49,32 +52,42 @@ abstract class FlussCatalogReadWriteITCaseBase extends PaimonSparkTestBase { override protected def sparkConf: SparkConf = { val conf = super.sparkConf - val bootstrapServers = flussCluster.getBootstrapServers - val paimonExtensions = conf.get("spark.sql.extensions") - - conf - .set(s"spark.sql.catalog.$flussCatalogName", "org.apache.fluss.spark.SparkCatalog") - .set(s"spark.sql.catalog.$flussCatalogName.bootstrap.servers", bootstrapServers) - .set(s"spark.sql.catalog.$flussCatalogName.datalake.format", "paimon") - .set(s"spark.sql.catalog.$flussCatalogName.datalake.paimon.metastore", "filesystem") - .set(s"spark.sql.catalog.$flussCatalogName.datalake.paimon.warehouse", lakeWarehouse.toString) - .set("spark.sql.catalog.paimon.warehouse", lakeWarehouse.toString) - .set("spark.sql.catalog.paimon.fluss.bootstrap.servers", bootstrapServers) - .set("spark.sql.catalog.paimon.fluss.datalake.format", "paimon") - .set("spark.sql.catalog.paimon.fluss.datalake.paimon.metastore", "filesystem") - .set("spark.sql.catalog.paimon.fluss.datalake.paimon.warehouse", lakeWarehouse.toString) - .set( - "spark.sql.extensions", - s"$paimonExtensions,org.apache.fluss.spark.FlussSparkSessionExtensions") + if (flussSparkConnectorAvailable) { + val bootstrapServers = flussCluster.getBootstrapServers + val paimonExtensions = conf.get("spark.sql.extensions") + + conf + .set(s"spark.sql.catalog.$flussCatalogName", "org.apache.fluss.spark.SparkCatalog") + .set(s"spark.sql.catalog.$flussCatalogName.bootstrap.servers", bootstrapServers) + .set(s"spark.sql.catalog.$flussCatalogName.datalake.format", "paimon") + .set(s"spark.sql.catalog.$flussCatalogName.datalake.paimon.metastore", "filesystem") + .set( + s"spark.sql.catalog.$flussCatalogName.datalake.paimon.warehouse", + lakeWarehouse.toString) + .set("spark.sql.catalog.paimon.warehouse", lakeWarehouse.toString) + .set("spark.sql.catalog.paimon.fluss.bootstrap.servers", bootstrapServers) + .set("spark.sql.catalog.paimon.fluss.datalake.format", "paimon") + .set("spark.sql.catalog.paimon.fluss.datalake.paimon.metastore", "filesystem") + .set("spark.sql.catalog.paimon.fluss.datalake.paimon.warehouse", lakeWarehouse.toString) + .set( + "spark.sql.extensions", + s"$paimonExtensions,org.apache.fluss.spark.FlussSparkSessionExtensions") + } else { + conf + } } override protected def beforeAll(): Unit = { - flussCluster.start() + if (flussSparkConnectorAvailable) { + flussCluster.start() + } try { super.beforeAll() } catch { case t: Throwable => - flussCluster.close() + if (flussSparkConnectorAvailable) { + flussCluster.close() + } throw t } } @@ -84,7 +97,9 @@ abstract class FlussCatalogReadWriteITCaseBase extends PaimonSparkTestBase { super.afterAll() } finally { try { - flussCluster.close() + if (flussSparkConnectorAvailable) { + flussCluster.close() + } } finally { FileIOUtils.deleteDirectoryQuietly(lakeWarehouse.getParent.toFile) } @@ -92,6 +107,7 @@ abstract class FlussCatalogReadWriteITCaseBase extends PaimonSparkTestBase { } test("read and write a Fluss log table through Paimon catalog") { + assume(flussSparkConnectorAvailable) withFlussTable("log_orders") { verifyLakeStreamMarker("log_orders") @@ -108,6 +124,7 @@ abstract class FlussCatalogReadWriteITCaseBase extends PaimonSparkTestBase { } test("read and write a Fluss primary-key table through Paimon catalog") { + assume(flussSparkConnectorAvailable) withFlussTable("pk_orders", primaryKey = true) { verifyLakeStreamMarker("pk_orders") diff --git a/pom.xml b/pom.xml index c41481e66195..92555a3c8398 100644 --- a/pom.xml +++ b/pom.xml @@ -114,6 +114,7 @@ under the License. 1.19.1 1.6.1 1.0-SNAPSHOT + 2.12 5.4.0 1.16.0 1.9.8 From dd31c2da9f5a40e342119b239efe04bc0c7b6a15 Mon Sep 17 00:00:00 2001 From: zouxxyy Date: Thu, 3 Sep 2026 08:56:23 +0800 Subject: [PATCH 3/6] [spark] Run Fluss integration tests with Java 11 Keep Java 11 Fluss test artifacts off the Spark 3 Java 8 Surefire classpath and avoid initializing the MiniCluster below Java 11. Verify the optional connector behavior and run the Fluss read/write suites in the existing JDK 11 workflow. Co-Authored-By: Claude Code Co-Authored-By: Codex AI-Model: gpt-5 AI-Contributed/Feature: 34/34 AI-Contributed/UT: 53/53 --- .github/workflows/utitcase-jdk11.yml | 8 ++++++ .../FlussCatalogReadWriteITCaseBase.scala | 25 +++++++++++------ .../paimon/spark/FlussCatalogTestBase.scala | 28 ++++++++++++++++--- paimon-spark/pom.xml | 26 +++++++++++++++++ 4 files changed, 74 insertions(+), 13 deletions(-) diff --git a/.github/workflows/utitcase-jdk11.yml b/.github/workflows/utitcase-jdk11.yml index a6ae09a52be2..cd0a002d3752 100644 --- a/.github/workflows/utitcase-jdk11.yml +++ b/.github/workflows/utitcase-jdk11.yml @@ -56,6 +56,14 @@ jobs: distribution: 'temurin' - name: Build run: mvn -T 1C -B -ntp clean install -DskipTests -Pflink1,spark3 + - name: Test Fluss with Spark 3 + run: | + mvn -B -ntp verify \ + -pl org.apache.paimon:paimon-spark-3.5_2.12,org.apache.paimon:paimon-spark-3.4_2.12 \ + -Pflink1,spark3 \ + -DfailIfNoTests=false \ + -Dtest=none \ + -DwildcardSuites=org.apache.paimon.spark.FlussCatalogTest,org.apache.paimon.spark.FlussCatalogReadWriteITCase - name: Test run: | # run tests with random timezone to find out timezone related bugs diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/FlussCatalogReadWriteITCaseBase.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/FlussCatalogReadWriteITCaseBase.scala index a75ea2b943aa..8b77c9a2047f 100644 --- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/FlussCatalogReadWriteITCaseBase.scala +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/FlussCatalogReadWriteITCaseBase.scala @@ -34,15 +34,15 @@ import java.time.Duration /** MiniCluster tests for reading and writing Fluss LakeStream tables through Paimon catalog. */ abstract class FlussCatalogReadWriteITCaseBase extends PaimonSparkTestBase { - private val flussSparkConnectorAvailable = - scala.util.Properties.versionNumberString.startsWith("2.12.") + private val flussTestEnvironmentAvailable = + scala.util.Properties.versionNumberString.startsWith("2.12.") && javaMajorVersion >= 11 private val flussCatalogName = "fluss_catalog" private val flussDatabase = "fluss" private val lakeWarehouse: Path = Files.createTempDirectory("paimon-fluss-catalog-it").resolve("warehouse") - private val flussCluster: FlussClusterExtension = + private lazy val flussCluster: FlussClusterExtension = FlussClusterExtension.builder .setClusterConf(flussConfiguration) .setNumOfTabletServers(1) @@ -52,7 +52,7 @@ abstract class FlussCatalogReadWriteITCaseBase extends PaimonSparkTestBase { override protected def sparkConf: SparkConf = { val conf = super.sparkConf - if (flussSparkConnectorAvailable) { + if (flussTestEnvironmentAvailable) { val bootstrapServers = flussCluster.getBootstrapServers val paimonExtensions = conf.get("spark.sql.extensions") @@ -78,14 +78,14 @@ abstract class FlussCatalogReadWriteITCaseBase extends PaimonSparkTestBase { } override protected def beforeAll(): Unit = { - if (flussSparkConnectorAvailable) { + if (flussTestEnvironmentAvailable) { flussCluster.start() } try { super.beforeAll() } catch { case t: Throwable => - if (flussSparkConnectorAvailable) { + if (flussTestEnvironmentAvailable) { flussCluster.close() } throw t @@ -97,7 +97,7 @@ abstract class FlussCatalogReadWriteITCaseBase extends PaimonSparkTestBase { super.afterAll() } finally { try { - if (flussSparkConnectorAvailable) { + if (flussTestEnvironmentAvailable) { flussCluster.close() } } finally { @@ -107,7 +107,9 @@ abstract class FlussCatalogReadWriteITCaseBase extends PaimonSparkTestBase { } test("read and write a Fluss log table through Paimon catalog") { - assume(flussSparkConnectorAvailable) + assume( + flussTestEnvironmentAvailable, + "Fluss Spark integration tests require Scala 2.12 and Java 11 or later") withFlussTable("log_orders") { verifyLakeStreamMarker("log_orders") @@ -124,7 +126,9 @@ abstract class FlussCatalogReadWriteITCaseBase extends PaimonSparkTestBase { } test("read and write a Fluss primary-key table through Paimon catalog") { - assume(flussSparkConnectorAvailable) + assume( + flussTestEnvironmentAvailable, + "Fluss Spark integration tests require Scala 2.12 and Java 11 or later") withFlussTable("pk_orders", primaryKey = true) { verifyLakeStreamMarker("pk_orders") @@ -196,4 +200,7 @@ abstract class FlussCatalogReadWriteITCaseBase extends PaimonSparkTestBase { conf.setString("server.data-disk.write-limit-ratio", "1.0") conf } + + private def javaMajorVersion: Int = + System.getProperty("java.specification.version").split("\\.").last.toInt } diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/FlussCatalogTestBase.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/FlussCatalogTestBase.scala index 742df3a0a61f..c46132ee21f2 100644 --- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/FlussCatalogTestBase.scala +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/FlussCatalogTestBase.scala @@ -90,13 +90,23 @@ abstract class FlussCatalogTestBase extends PaimonSparkTestBase { assert(flussCatalog.initializeCount == 1) } - test("keep a regular file store table on Paimon") { + test("keep a regular file store table on Paimon without loading Fluss") { val paimonTable = SparkTable(fileStoreTable(Collections.emptyMap[String, String]())) - val flussCatalog = new TestingTableCatalog - val catalog = testingSparkCatalog(paimonTable, configuredDelegate(flussCatalog)) + val catalog = testingSparkCatalog(paimonTable, missingFlussDelegate) assert(catalog.loadTable(tableIdentifier) eq paimonTable) - assert(flussCatalog.initializeCount == 0) + } + + test("report a missing Fluss connector for a marked table") { + val paimonTable = + SparkTable(fileStoreTable(Collections.singletonMap(SupportFluss.LAKESTREAM_ENABLED, "true"))) + val catalog = testingSparkCatalog(paimonTable, missingFlussDelegate) + + val error = intercept[IllegalStateException] { + catalog.loadTable(tableIdentifier) + } + assert(error.getMessage.contains("matching the Spark version")) + assert(error.getCause.isInstanceOf[ClassNotFoundException]) } test("do not route a non-file-store table carrying the marker") { @@ -118,6 +128,16 @@ abstract class FlussCatalogTestBase extends PaimonSparkTestBase { } ) + private def missingFlussDelegate: FlussCatalogDelegate = + new FlussCatalogDelegate( + Collections.singletonMap("fluss.bootstrap.servers", "localhost:9123"), + "paimon", + new FlussCatalogDelegate.CatalogLoader { + override def load(classLoader: ClassLoader): TableCatalog = + throw new ClassNotFoundException("org.apache.fluss.spark.SparkCatalog") + } + ) + private def testingSparkCatalog(table: Table, delegate: FlussCatalogDelegate): SparkCatalog = { val catalog = new TestingSparkCatalog(table) val field: Field = classOf[SparkCatalog].getDeclaredField("flussCatalogDelegate") diff --git a/paimon-spark/pom.xml b/paimon-spark/pom.xml index 7e38786043c7..d96ed2c55b68 100644 --- a/paimon-spark/pom.xml +++ b/paimon-spark/pom.xml @@ -385,4 +385,30 @@ under the License. + + + + exclude-fluss-tests-on-java8 + + [1.8,1.9) + + + + + org.apache.maven.plugins + maven-surefire-plugin + + + + org.apache.fluss:fluss-common + org.apache.fluss:fluss-server + org.apache.fluss:fluss-test-utils + + + + + + + From 45ff3249fb6b945bc634f5ac82daafb84d27ee04 Mon Sep 17 00:00:00 2001 From: zouxxyy Date: Thu, 3 Sep 2026 10:37:00 +0800 Subject: [PATCH 4/6] [spark] Skip Fluss integration test discovery on Java 8 Keep the Java 11 Fluss test classes out of JUnit Vintage discovery after excluding their runtime dependencies from the Java 8 Surefire classpath. Co-Authored-By: Claude Code Co-Authored-By: Codex AI-Model: gpt-5 AI-Contributed/Feature: 13/13 AI-Contributed/UT: 0/0 --- paimon-spark/pom.xml | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/paimon-spark/pom.xml b/paimon-spark/pom.xml index d96ed2c55b68..3be537f55578 100644 --- a/paimon-spark/pom.xml +++ b/paimon-spark/pom.xml @@ -406,6 +406,19 @@ under the License. org.apache.fluss:fluss-test-utils + + + integration-tests + + + + **/FlussCatalogReadWriteITCase.* + + + + From 80443ed9b8d3f1ec5f8a7a5486b9fb70add81dd1 Mon Sep 17 00:00:00 2001 From: zouxxyy Date: Thu, 3 Sep 2026 12:27:52 +0800 Subject: [PATCH 5/6] [spark] Route LakeStream reads to Paimon and Fluss Read base Fluss LakeStream tables through Paimon while routing writes to Fluss. Expose $rt as a read-only Fluss union view and leave $stream unsupported. Keep catalog routing in spark.catalog and table adapters in spark. Co-Authored-By: Claude Code Co-Authored-By: Codex AI-Model: gpt-5 AI-Contributed/Feature: 279/279 AI-Contributed/UT: 196/196 --- .../spark/FlussLakeStreamReadTable.java | 74 +++++++ .../paimon/spark/FlussLakeStreamTable.java | 115 +++++++++++ .../org/apache/paimon/spark/SparkCatalog.java | 8 +- .../{ => catalog}/FlussCatalogDelegate.java | 24 +-- .../paimon/spark/catalog/SupportFluss.java | 58 +++++- .../FlussCatalogReadWriteITCaseBase.scala | 14 +- .../paimon/spark/FlussCatalogTestBase.scala | 182 ++++++++++++++++-- 7 files changed, 430 insertions(+), 45 deletions(-) create mode 100644 paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/FlussLakeStreamReadTable.java create mode 100644 paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/FlussLakeStreamTable.java rename paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/{ => catalog}/FlussCatalogDelegate.java (90%) diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/FlussLakeStreamReadTable.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/FlussLakeStreamReadTable.java new file mode 100644 index 000000000000..469e0f0114de --- /dev/null +++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/FlussLakeStreamReadTable.java @@ -0,0 +1,74 @@ +/* + * 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; + +import org.apache.spark.sql.connector.catalog.SupportsRead; +import org.apache.spark.sql.connector.catalog.Table; +import org.apache.spark.sql.connector.catalog.TableCapability; +import org.apache.spark.sql.connector.expressions.Transform; +import org.apache.spark.sql.connector.read.ScanBuilder; +import org.apache.spark.sql.types.StructType; +import org.apache.spark.sql.util.CaseInsensitiveStringMap; + +import java.util.Map; +import java.util.Set; + +/** A read-only Fluss LakeStream table used by the {@code $rt} suffix. */ +public final class FlussLakeStreamReadTable implements Table, SupportsRead { + + private final Table flussTable; + + public FlussLakeStreamReadTable(Table flussTable) { + if (!(flussTable instanceof SupportsRead)) { + throw new IllegalArgumentException( + String.format("Fluss table '%s' does not support reads.", flussTable.name())); + } + this.flussTable = flussTable; + } + + @Override + public String name() { + return flussTable.name(); + } + + @Override + public StructType schema() { + return flussTable.schema(); + } + + @Override + public Transform[] partitioning() { + return flussTable.partitioning(); + } + + @Override + public Map properties() { + return flussTable.properties(); + } + + @Override + public Set capabilities() { + return FlussLakeStreamTable.readCapabilities(flussTable); + } + + @Override + public ScanBuilder newScanBuilder(CaseInsensitiveStringMap options) { + return ((SupportsRead) flussTable).newScanBuilder(options); + } +} diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/FlussLakeStreamTable.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/FlussLakeStreamTable.java new file mode 100644 index 000000000000..090c0bd423ea --- /dev/null +++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/FlussLakeStreamTable.java @@ -0,0 +1,115 @@ +/* + * 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; + +import org.apache.spark.sql.connector.catalog.MetadataColumn; +import org.apache.spark.sql.connector.catalog.SupportsMetadataColumns; +import org.apache.spark.sql.connector.catalog.SupportsRead; +import org.apache.spark.sql.connector.catalog.SupportsWrite; +import org.apache.spark.sql.connector.catalog.Table; +import org.apache.spark.sql.connector.catalog.TableCapability; +import org.apache.spark.sql.connector.expressions.Transform; +import org.apache.spark.sql.connector.read.ScanBuilder; +import org.apache.spark.sql.connector.write.LogicalWriteInfo; +import org.apache.spark.sql.connector.write.WriteBuilder; +import org.apache.spark.sql.types.StructType; +import org.apache.spark.sql.util.CaseInsensitiveStringMap; + +import java.util.EnumSet; +import java.util.Map; +import java.util.Set; + +/** A Fluss LakeStream table which reads from Paimon and writes to Fluss. */ +public final class FlussLakeStreamTable + implements Table, SupportsRead, SupportsWrite, SupportsMetadataColumns { + + private final SparkTable paimonTable; + private final Table flussTable; + + public FlussLakeStreamTable(SparkTable paimonTable, Table flussTable) { + if (!(flussTable instanceof SupportsWrite)) { + throw new IllegalArgumentException( + String.format("Fluss table '%s' does not support writes.", flussTable.name())); + } + this.paimonTable = paimonTable; + this.flussTable = flussTable; + } + + @Override + public String name() { + return paimonTable.name(); + } + + @Override + public StructType schema() { + // Fluss owns the user-visible schema used to validate writes. The Paimon lake table may + // contain a different physical layout. + return flussTable.schema(); + } + + @Override + public Transform[] partitioning() { + return paimonTable.partitioning(); + } + + @Override + public Map properties() { + return paimonTable.properties(); + } + + @Override + public Set capabilities() { + Set capabilities = readCapabilities(paimonTable); + addIfSupported(capabilities, flussTable, TableCapability.BATCH_WRITE); + addIfSupported(capabilities, flussTable, TableCapability.STREAMING_WRITE); + return capabilities; + } + + @Override + public MetadataColumn[] metadataColumns() { + return paimonTable.metadataColumns(); + } + + @Override + public ScanBuilder newScanBuilder(CaseInsensitiveStringMap options) { + return paimonTable.newScanBuilder(options); + } + + @Override + public WriteBuilder newWriteBuilder(LogicalWriteInfo info) { + return ((SupportsWrite) flussTable).newWriteBuilder(info); + } + + static Set readCapabilities(Table table) { + Set capabilities = EnumSet.noneOf(TableCapability.class); + for (TableCapability capability : table.capabilities()) { + if (capability.name().endsWith("_READ")) { + capabilities.add(capability); + } + } + return capabilities; + } + + private static void addIfSupported( + Set target, Table source, TableCapability capability) { + if (source.capabilities().contains(capability)) { + target.add(capability); + } + } +} diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java index ee224dbaf30a..9b2e95ca84f1 100644 --- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java +++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java @@ -30,6 +30,7 @@ import org.apache.paimon.rest.RESTCatalog; import org.apache.paimon.schema.Schema; import org.apache.paimon.schema.SchemaChange; +import org.apache.paimon.spark.catalog.FlussCatalogDelegate; import org.apache.paimon.spark.catalog.FormatTableCatalog; import org.apache.paimon.spark.catalog.SparkBaseCatalog; import org.apache.paimon.spark.catalog.SupportFluss; @@ -313,9 +314,10 @@ public void invalidateTable(Identifier ident) { @Override public org.apache.spark.sql.connector.catalog.Table loadTable(Identifier ident) throws NoSuchTableException { - org.apache.spark.sql.connector.catalog.Table table = - loadSparkTable(ident, Collections.emptyMap()); - return isFlussTable(table) ? flussCatalogDelegate.loadTable(ident) : table; + return loadTableWithFluss( + ident, + identifier -> loadSparkTable(identifier, Collections.emptyMap()), + flussCatalogDelegate::loadTable); } /** diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/FlussCatalogDelegate.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/FlussCatalogDelegate.java similarity index 90% rename from paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/FlussCatalogDelegate.java rename to paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/FlussCatalogDelegate.java index 27ff9d2a982b..a8b082bba31e 100644 --- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/FlussCatalogDelegate.java +++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/FlussCatalogDelegate.java @@ -16,7 +16,9 @@ * limitations under the License. */ -package org.apache.paimon.spark; +package org.apache.paimon.spark.catalog; + +import org.apache.paimon.annotation.VisibleForTesting; import org.apache.spark.sql.catalyst.analysis.NoSuchTableException; import org.apache.spark.sql.connector.catalog.Identifier; @@ -24,12 +26,11 @@ import org.apache.spark.sql.connector.catalog.TableCatalog; import org.apache.spark.sql.util.CaseInsensitiveStringMap; -import java.util.HashMap; import java.util.Map; import java.util.stream.Collectors; /** Loads LakeStream tables from the Fluss Spark catalog used by Spark reads and writes. */ -class FlussCatalogDelegate { +public final class FlussCatalogDelegate { static final String FLUSS_OPTION_PREFIX = "fluss."; static final String FLUSS_BOOTSTRAP_SERVERS = "fluss.bootstrap.servers"; @@ -42,15 +43,17 @@ class FlussCatalogDelegate { private volatile TableCatalog flussCatalog; - FlussCatalogDelegate(Map catalogOptions, String catalogName) { + public FlussCatalogDelegate(Map catalogOptions, String catalogName) { this(catalogOptions, catalogName, FlussCatalogDelegate::loadFlussCatalog); } - FlussCatalogDelegate( + @VisibleForTesting + public FlussCatalogDelegate( Map catalogOptions, String catalogName, CatalogLoader catalogLoader) { this.catalogName = catalogName; + CaseInsensitiveStringMap options = new CaseInsensitiveStringMap(catalogOptions); this.flussOptions = - catalogOptions.entrySet().stream() + options.entrySet().stream() .filter(entry -> entry.getKey().startsWith(FLUSS_OPTION_PREFIX)) .collect( Collectors.toMap( @@ -65,11 +68,7 @@ boolean isConfigured() { return flussOptions.containsKey("bootstrap.servers"); } - Map flussOptions() { - return new HashMap<>(flussOptions); - } - - Table loadTable(Identifier identifier) throws NoSuchTableException { + public Table loadTable(Identifier identifier) throws NoSuchTableException { return catalog().loadTable(identifier); } @@ -120,7 +119,8 @@ private static ClassLoader contextClassLoader() { return classLoader == null ? FlussCatalogDelegate.class.getClassLoader() : classLoader; } - interface CatalogLoader { + @VisibleForTesting + public interface CatalogLoader { TableCatalog load(ClassLoader classLoader) throws Exception; } } diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/SupportFluss.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/SupportFluss.java index 7d79227454ab..a38f7f9176bd 100644 --- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/SupportFluss.java +++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/SupportFluss.java @@ -18,25 +18,73 @@ package org.apache.paimon.spark.catalog; +import org.apache.paimon.spark.FlussLakeStreamReadTable; +import org.apache.paimon.spark.FlussLakeStreamTable; import org.apache.paimon.spark.SparkTable; import org.apache.paimon.table.FileStoreTable; +import org.apache.spark.sql.catalyst.analysis.NoSuchTableException; +import org.apache.spark.sql.connector.catalog.Identifier; import org.apache.spark.sql.connector.catalog.Table; +import java.util.Optional; + /** Catalog support for Fluss LakeStream tables backed by Paimon. */ public interface SupportFluss { String LAKESTREAM_ENABLED = "lakestream.enabled"; + String REAL_TIME_SUFFIX = "$rt"; + + /** Loads a Paimon table, routing LakeStream writes and {@code $rt} reads through Fluss. */ + default Table loadTableWithFluss( + Identifier identifier, TableLoader paimonTableLoader, TableLoader flussTableLoader) + throws NoSuchTableException { + Optional realTimeTableBase = flussLakeStreamBaseIdentifier(identifier); + if (realTimeTableBase.isPresent()) { + Identifier baseIdentifier = realTimeTableBase.get(); + Table baseTable; + try { + baseTable = paimonTableLoader.load(baseIdentifier); + } catch (NoSuchTableException e) { + throw new NoSuchTableException(identifier); + } + if (!isFlussLakeStreamTable(baseTable)) { + throw new NoSuchTableException(identifier); + } + return new FlussLakeStreamReadTable(flussTableLoader.load(baseIdentifier)); + } + + Table paimonTable = paimonTableLoader.load(identifier); + return isFlussLakeStreamTable(paimonTable) + ? new FlussLakeStreamTable( + (SparkTable) paimonTable, flussTableLoader.load(identifier)) + : paimonTable; + } - default boolean isFlussTable(Table table) { + default boolean isFlussLakeStreamTable(Table table) { if (!(table instanceof SparkTable)) { return false; } - return isFlussTable(((SparkTable) table).getTable()); + org.apache.paimon.table.Table paimonTable = ((SparkTable) table).getTable(); + return paimonTable instanceof FileStoreTable + && Boolean.parseBoolean(paimonTable.options().get(LAKESTREAM_ENABLED)); + } + + /** Returns the base identifier of a Fluss {@code $rt} table, if present. */ + default Optional flussLakeStreamBaseIdentifier(Identifier identifier) { + String tableName = identifier.name(); + int suffixStart = tableName.length() - REAL_TIME_SUFFIX.length(); + if (suffixStart <= 0 + || !tableName.endsWith(REAL_TIME_SUFFIX) + || tableName.indexOf('$') != suffixStart) { + return Optional.empty(); + } + return Optional.of( + Identifier.of(identifier.namespace(), tableName.substring(0, suffixStart))); } - default boolean isFlussTable(org.apache.paimon.table.Table table) { - return table instanceof FileStoreTable - && Boolean.parseBoolean(table.options().get(LAKESTREAM_ENABLED)); + @FunctionalInterface + interface TableLoader { + Table load(Identifier identifier) throws NoSuchTableException; } } diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/FlussCatalogReadWriteITCaseBase.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/FlussCatalogReadWriteITCaseBase.scala index 8b77c9a2047f..f660d1354fca 100644 --- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/FlussCatalogReadWriteITCaseBase.scala +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/FlussCatalogReadWriteITCaseBase.scala @@ -106,7 +106,7 @@ abstract class FlussCatalogReadWriteITCaseBase extends PaimonSparkTestBase { } } - test("read and write a Fluss log table through Paimon catalog") { + test("write a Fluss log table and read its lake and real-time views") { assume( flussTestEnvironmentAvailable, "Fluss Spark integration tests require Scala 2.12 and Java 11 or later") @@ -114,18 +114,19 @@ abstract class FlussCatalogReadWriteITCaseBase extends PaimonSparkTestBase { verifyLakeStreamMarker("log_orders") sql("INSERT INTO paimon.fluss.log_orders VALUES (1, 'a'), (2, 'b')") + checkAnswer(sql("SELECT * FROM paimon.fluss.log_orders"), Nil) checkAnswer( - sql("SELECT * FROM fluss_catalog.fluss.log_orders"), + sql("SELECT * FROM paimon.fluss.`log_orders$rt`"), Row(1, "a") :: Row(2, "b") :: Nil) sql("INSERT INTO fluss_catalog.fluss.log_orders VALUES (3, 'c')") checkAnswer( - sql("SELECT * FROM paimon.fluss.log_orders"), + sql("SELECT * FROM paimon.fluss.`log_orders$rt`"), Row(1, "a") :: Row(2, "b") :: Row(3, "c") :: Nil) } } - test("read and write a Fluss primary-key table through Paimon catalog") { + test("write a Fluss primary-key table and read its lake and real-time views") { assume( flussTestEnvironmentAvailable, "Fluss Spark integration tests require Scala 2.12 and Java 11 or later") @@ -133,13 +134,14 @@ abstract class FlussCatalogReadWriteITCaseBase extends PaimonSparkTestBase { verifyLakeStreamMarker("pk_orders") sql("INSERT INTO paimon.fluss.pk_orders VALUES (1, 'a'), (2, 'b')") + checkAnswer(sql("SELECT * FROM paimon.fluss.pk_orders"), Nil) checkAnswer( - sql("SELECT * FROM fluss_catalog.fluss.pk_orders"), + sql("SELECT * FROM paimon.fluss.`pk_orders$rt`"), Row(1, "a") :: Row(2, "b") :: Nil) sql("INSERT INTO fluss_catalog.fluss.pk_orders VALUES (1, 'a2'), (3, 'c')") checkAnswer( - sql("SELECT * FROM paimon.fluss.pk_orders"), + sql("SELECT * FROM paimon.fluss.`pk_orders$rt`"), Row(1, "a2") :: Row(2, "b") :: Row(3, "c") :: Nil) } } diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/FlussCatalogTestBase.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/FlussCatalogTestBase.scala index c46132ee21f2..91aec26bfc27 100644 --- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/FlussCatalogTestBase.scala +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/FlussCatalogTestBase.scala @@ -18,16 +18,18 @@ package org.apache.paimon.spark -import org.apache.paimon.spark.catalog.SupportFluss +import org.apache.paimon.spark.catalog.{FlussCatalogDelegate, SupportFluss} import org.apache.paimon.table.FileStoreTable -import org.apache.spark.sql.connector.catalog.{Identifier, Table, TableCapability, TableCatalog, TableChange} +import org.apache.spark.sql.connector.catalog.{Identifier, MetadataColumn, SupportsRead, SupportsWrite, Table, TableCapability, TableCatalog, TableChange, TruncatableTable} import org.apache.spark.sql.connector.expressions.Transform -import org.apache.spark.sql.types.StructType +import org.apache.spark.sql.connector.read.{Scan, ScanBuilder} +import org.apache.spark.sql.connector.write.{LogicalWriteInfo, Write, WriteBuilder} +import org.apache.spark.sql.types.{DataTypes, StructType} import org.apache.spark.sql.util.CaseInsensitiveStringMap import java.lang.reflect.{Field, InvocationHandler, Method, Proxy} -import java.util.{Collections, HashMap, Map => JMap, Set => JSet} +import java.util.{Collections, EnumSet, HashMap, Map => JMap, Set => JSet} import java.util.concurrent.atomic.AtomicReference import scala.collection.JavaConverters._ @@ -36,6 +38,9 @@ import scala.collection.JavaConverters._ abstract class FlussCatalogTestBase extends PaimonSparkTestBase { private val tableIdentifier = Identifier.of(Array("db"), "orders") + private val realTimeIdentifier = Identifier.of(Array("db"), "orders$rt") + private val systemTableIdentifier = Identifier.of(Array("db"), "orders$snapshots") + private val emptyOptions = new CaseInsensitiveStringMap(Collections.emptyMap[String, String]()) test("map Fluss options and initialize the delegate lazily") { val catalog = new TestingTableCatalog @@ -43,7 +48,7 @@ abstract class FlussCatalogTestBase extends PaimonSparkTestBase { val options = new HashMap[String, String] options.put("warehouse", "/tmp/warehouse") options.put("fluss.bootstrap.servers", "localhost:9123") - options.put("fluss.client.security.protocol", "sasl") + options.put("FLUSS.client.security.protocol", "sasl") val delegate = new FlussCatalogDelegate( options, "paimon", @@ -55,10 +60,6 @@ abstract class FlussCatalogTestBase extends PaimonSparkTestBase { } ) - assert( - delegate.flussOptions().asScala == Map( - "bootstrap.servers" -> "localhost:9123", - "client.security.protocol" -> "sasl")) assert(delegate.loadTable(tableIdentifier) eq catalog.table) assert(catalog.catalogName == "paimon") assert( @@ -80,18 +81,90 @@ abstract class FlussCatalogTestBase extends PaimonSparkTestBase { assert(error.getMessage.contains("fluss.bootstrap.servers")) } - test("route a marked file store table to Fluss") { + test("read a marked base table from Paimon and write it to Fluss") { val paimonTable = - SparkTable(fileStoreTable(Collections.singletonMap(SupportFluss.LAKESTREAM_ENABLED, "true"))) + new TestingSparkTable( + fileStoreTable(Collections.singletonMap(SupportFluss.LAKESTREAM_ENABLED, "true"))) val flussCatalog = new TestingTableCatalog val catalog = testingSparkCatalog(paimonTable, configuredDelegate(flussCatalog)) - assert(catalog.loadTable(tableIdentifier) eq flussCatalog.table) + val table = catalog.loadTable(tableIdentifier) + assert(table.isInstanceOf[FlussLakeStreamTable]) + assert(!table.isInstanceOf[SparkTable]) + assert(!table.isInstanceOf[TruncatableTable]) + assert(table.schema() eq flussCatalog.table.schema()) + assert(table.properties().get(SupportFluss.LAKESTREAM_ENABLED) == "true") + assert(table.asInstanceOf[SupportsRead].newScanBuilder(emptyOptions) eq paimonTable.scanBuilder) + assert( + table.asInstanceOf[SupportsWrite].newWriteBuilder(null) eq flussCatalog.table.writeBuilder) + assert(table.capabilities().contains(TableCapability.BATCH_READ)) + assert(table.capabilities().contains(TableCapability.BATCH_WRITE)) + assert(!table.capabilities().contains(TableCapability.V1_BATCH_WRITE)) assert(flussCatalog.initializeCount == 1) + assert(flussCatalog.lastIdentifier == tableIdentifier) + } + + test("route the $rt suffix to a read-only Fluss table") { + val paimonTable = + new TestingSparkTable( + fileStoreTable(Collections.singletonMap(SupportFluss.LAKESTREAM_ENABLED, "true"))) + val flussCatalog = new TestingTableCatalog + val catalog = testingSparkCatalog(paimonTable, configuredDelegate(flussCatalog)) + + val table = catalog.loadTable(realTimeIdentifier) + assert(table.isInstanceOf[FlussLakeStreamReadTable]) + assert(table.isInstanceOf[SupportsRead]) + assert(!table.isInstanceOf[SupportsWrite]) + assert( + table + .asInstanceOf[SupportsRead] + .newScanBuilder(emptyOptions) eq flussCatalog.table.scanBuilder) + assert(flussCatalog.lastIdentifier == tableIdentifier) + assert(catalog.lastPaimonIdentifier == tableIdentifier) + } + + test("reject $rt for a regular Paimon table") { + val paimonTable = + new TestingSparkTable(fileStoreTable(Collections.emptyMap[String, String]())) + val flussCatalog = new TestingTableCatalog + val catalog = testingSparkCatalog(paimonTable, configuredDelegate(flussCatalog)) + + intercept[org.apache.spark.sql.catalyst.analysis.NoSuchTableException] { + catalog.loadTable(realTimeIdentifier) + } + assert(flussCatalog.initializeCount == 0) + } + + test("keep Paimon system tables on Paimon") { + val paimonTable = + new TestingSparkTable( + fileStoreTable(Collections.singletonMap(SupportFluss.LAKESTREAM_ENABLED, "true"))) + val systemTable = new TestingTable(Collections.emptyMap[String, String]()) + val flussCatalog = new TestingTableCatalog + val catalog = + testingSparkCatalog(paimonTable, configuredDelegate(flussCatalog), systemTable) + + assert(catalog.loadTable(systemTableIdentifier) eq systemTable) + assert(catalog.lastPaimonIdentifier == systemTableIdentifier) + assert(flussCatalog.initializeCount == 0) + } + + test("do not support the $stream suffix") { + val paimonTable = + new TestingSparkTable( + fileStoreTable(Collections.singletonMap(SupportFluss.LAKESTREAM_ENABLED, "true"))) + val flussCatalog = new TestingTableCatalog + val catalog = testingSparkCatalog(paimonTable, configuredDelegate(flussCatalog)) + + intercept[org.apache.spark.sql.catalyst.analysis.NoSuchTableException] { + catalog.loadTable(Identifier.of(Array("db"), "orders$stream")) + } + assert(flussCatalog.initializeCount == 0) } test("keep a regular file store table on Paimon without loading Fluss") { - val paimonTable = SparkTable(fileStoreTable(Collections.emptyMap[String, String]())) + val paimonTable = + new TestingSparkTable(fileStoreTable(Collections.emptyMap[String, String]())) val catalog = testingSparkCatalog(paimonTable, missingFlussDelegate) assert(catalog.loadTable(tableIdentifier) eq paimonTable) @@ -99,7 +172,8 @@ abstract class FlussCatalogTestBase extends PaimonSparkTestBase { test("report a missing Fluss connector for a marked table") { val paimonTable = - SparkTable(fileStoreTable(Collections.singletonMap(SupportFluss.LAKESTREAM_ENABLED, "true"))) + new TestingSparkTable( + fileStoreTable(Collections.singletonMap(SupportFluss.LAKESTREAM_ENABLED, "true"))) val catalog = testingSparkCatalog(paimonTable, missingFlussDelegate) val error = intercept[IllegalStateException] { @@ -138,8 +212,11 @@ abstract class FlussCatalogTestBase extends PaimonSparkTestBase { } ) - private def testingSparkCatalog(table: Table, delegate: FlussCatalogDelegate): SparkCatalog = { - val catalog = new TestingSparkCatalog(table) + private def testingSparkCatalog( + table: Table, + delegate: FlussCatalogDelegate, + systemTable: Table = null): TestingSparkCatalog = { + val catalog = new TestingSparkCatalog(table, systemTable) val field: Field = classOf[SparkCatalog].getDeclaredField("flussCatalogDelegate") field.setAccessible(true) field.set(catalog, delegate) @@ -162,11 +239,22 @@ abstract class FlussCatalogTestBase extends PaimonSparkTestBase { ) .asInstanceOf[FileStoreTable] - private class TestingSparkCatalog(table: Table) extends SparkCatalog { + private class TestingSparkCatalog(table: Table, systemTable: Table) extends SparkCatalog { + + var lastPaimonIdentifier: Identifier = _ override protected def loadSparkTable( ident: Identifier, - extraOptions: JMap[String, String]): Table = table + extraOptions: JMap[String, String]): Table = { + lastPaimonIdentifier = ident + if (ident == tableIdentifier) { + table + } else if (ident == systemTableIdentifier && systemTable != null) { + systemTable + } else { + throw new org.apache.spark.sql.catalyst.analysis.NoSuchTableException(ident) + } + } } private class TestingTable(tableProperties: JMap[String, String]) extends Table { @@ -181,9 +269,65 @@ abstract class FlussCatalogTestBase extends PaimonSparkTestBase { override def properties(): JMap[String, String] = tableProperties } + private class TestingSparkTable(override val table: FileStoreTable) extends SparkTable(table) { + + private val tableSchema = + new StructType().add("lake_only", DataTypes.IntegerType, true) + + val scanBuilder: ScanBuilder = new ScanBuilder { + override def build(): Scan = null + } + + override def name(): String = "orders" + + override lazy val schema: StructType = tableSchema + + override def partitioning(): Array[Transform] = Array.empty + + override def properties(): JMap[String, String] = table.options() + + override def capabilities(): JSet[TableCapability] = + EnumSet.of( + TableCapability.BATCH_READ, + TableCapability.MICRO_BATCH_READ, + TableCapability.V1_BATCH_WRITE) + + override def metadataColumns(): Array[MetadataColumn] = Array.empty + + override def newScanBuilder(options: CaseInsensitiveStringMap): ScanBuilder = scanBuilder + } + + private class TestingReadWriteTable + extends TestingTable(Collections.emptyMap[String, String]()) + with SupportsRead + with SupportsWrite { + + private val tableSchema = new StructType().add("id", DataTypes.IntegerType, true) + + val scanBuilder: ScanBuilder = new ScanBuilder { + override def build(): Scan = null + } + val writeBuilder: WriteBuilder = new WriteBuilder { + override def build(): Write = null + } + + override def capabilities(): JSet[TableCapability] = + EnumSet.of( + TableCapability.BATCH_READ, + TableCapability.MICRO_BATCH_READ, + TableCapability.BATCH_WRITE, + TableCapability.STREAMING_WRITE) + + override def schema(): StructType = tableSchema + + override def newScanBuilder(options: CaseInsensitiveStringMap): ScanBuilder = scanBuilder + + override def newWriteBuilder(info: LogicalWriteInfo): WriteBuilder = writeBuilder + } + private class TestingTableCatalog extends TableCatalog { - val table: Table = new TestingTable(Collections.emptyMap[String, String]()) + val table = new TestingReadWriteTable var catalogName: String = _ var options: CaseInsensitiveStringMap = _ From a2664fac074da2f73452ef3d39c3a98de37c28d8 Mon Sep 17 00:00:00 2001 From: zouxxyy Date: Thu, 3 Sep 2026 14:00:46 +0800 Subject: [PATCH 6/6] [spark] Fix Fluss catalog checkstyle Document the loader interfaces required by Checkstyle. Co-Authored-By: Claude Code Co-Authored-By: Codex AI-Model: gpt-5 AI-Contributed/Feature: 2/2 AI-Contributed/UT: 0/0 --- .../org/apache/paimon/spark/catalog/FlussCatalogDelegate.java | 1 + .../main/java/org/apache/paimon/spark/catalog/SupportFluss.java | 1 + 2 files changed, 2 insertions(+) diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/FlussCatalogDelegate.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/FlussCatalogDelegate.java index a8b082bba31e..af8d606a2392 100644 --- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/FlussCatalogDelegate.java +++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/FlussCatalogDelegate.java @@ -119,6 +119,7 @@ private static ClassLoader contextClassLoader() { return classLoader == null ? FlussCatalogDelegate.class.getClassLoader() : classLoader; } + /** Loads a Fluss {@link TableCatalog} with the supplied class loader. */ @VisibleForTesting public interface CatalogLoader { TableCatalog load(ClassLoader classLoader) throws Exception; diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/SupportFluss.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/SupportFluss.java index a38f7f9176bd..0928f84c9d33 100644 --- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/SupportFluss.java +++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/SupportFluss.java @@ -83,6 +83,7 @@ default Optional flussLakeStreamBaseIdentifier(Identifier identifier Identifier.of(identifier.namespace(), tableName.substring(0, suffixStart))); } + /** Loads a Spark table for the given identifier. */ @FunctionalInterface interface TableLoader { Table load(Identifier identifier) throws NoSuchTableException;