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-3.4/pom.xml b/paimon-spark/paimon-spark-3.4/pom.xml index ffa65a71dba2..4eb4b66080dd 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_${fluss.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..32492ac1dbf0 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_${fluss.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/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 6fe7ea50335e..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,8 +30,10 @@ 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; import org.apache.paimon.spark.catalog.SupportV1Function; import org.apache.paimon.spark.catalog.SupportView; import org.apache.paimon.spark.catalog.functions.FunctionIdentifierConverter; @@ -117,6 +119,7 @@ /** Spark {@link TableCatalog} for paimon. */ public class SparkCatalog extends SparkBaseCatalog implements SupportView, + SupportFluss, SupportV1Function, FunctionCatalog, SupportsNamespaces, @@ -130,6 +133,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 +141,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 +314,10 @@ public void invalidateTable(Identifier ident) { @Override public org.apache.spark.sql.connector.catalog.Table loadTable(Identifier ident) throws NoSuchTableException { - return loadSparkTable(ident, Collections.emptyMap()); + 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/catalog/FlussCatalogDelegate.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/FlussCatalogDelegate.java new file mode 100644 index 000000000000..af8d606a2392 --- /dev/null +++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/FlussCatalogDelegate.java @@ -0,0 +1,127 @@ +/* + * 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.annotation.VisibleForTesting; + +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.Map; +import java.util.stream.Collectors; + +/** Loads LakeStream tables from the Fluss Spark catalog used by Spark reads and writes. */ +public final 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; + + public FlussCatalogDelegate(Map catalogOptions, String catalogName) { + this(catalogOptions, catalogName, FlussCatalogDelegate::loadFlussCatalog); + } + + @VisibleForTesting + public FlussCatalogDelegate( + Map catalogOptions, String catalogName, CatalogLoader catalogLoader) { + this.catalogName = catalogName; + CaseInsensitiveStringMap options = new CaseInsensitiveStringMap(catalogOptions); + this.flussOptions = + options.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"); + } + + public 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; + } + + /** 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 new file mode 100644 index 000000000000..0928f84c9d33 --- /dev/null +++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/SupportFluss.java @@ -0,0 +1,91 @@ +/* + * 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.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 isFlussLakeStreamTable(Table table) { + if (!(table instanceof SparkTable)) { + return false; + } + 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))); + } + + /** Loads a Spark table for the given identifier. */ + @FunctionalInterface + interface TableLoader { + Table load(Identifier identifier) throws NoSuchTableException; + } +} 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..f660d1354fca --- /dev/null +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/FlussCatalogReadWriteITCaseBase.scala @@ -0,0 +1,208 @@ +/* + * 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 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 lazy 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 + if (flussTestEnvironmentAvailable) { + 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 = { + if (flussTestEnvironmentAvailable) { + flussCluster.start() + } + try { + super.beforeAll() + } catch { + case t: Throwable => + if (flussTestEnvironmentAvailable) { + flussCluster.close() + } + throw t + } + } + + override protected def afterAll(): Unit = { + try { + super.afterAll() + } finally { + try { + if (flussTestEnvironmentAvailable) { + flussCluster.close() + } + } finally { + FileIOUtils.deleteDirectoryQuietly(lakeWarehouse.getParent.toFile) + } + } + } + + 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") + withFlussTable("log_orders") { + 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 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$rt`"), + Row(1, "a") :: Row(2, "b") :: Row(3, "c") :: Nil) + } + } + + 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") + withFlussTable("pk_orders", primaryKey = true) { + 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 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$rt`"), + 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 + } + + 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 new file mode 100644 index 000000000000..91aec26bfc27 --- /dev/null +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/FlussCatalogTestBase.scala @@ -0,0 +1,369 @@ +/* + * 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.{FlussCatalogDelegate, SupportFluss} +import org.apache.paimon.table.FileStoreTable + +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.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, EnumSet, 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") + 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 + 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.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("read a marked base table from Paimon and write it to Fluss") { + 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(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 = + new TestingSparkTable(fileStoreTable(Collections.emptyMap[String, String]())) + val catalog = testingSparkCatalog(paimonTable, missingFlussDelegate) + + assert(catalog.loadTable(tableIdentifier) eq paimonTable) + } + + test("report a missing Fluss connector for a marked table") { + val paimonTable = + new TestingSparkTable( + 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") { + 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 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, + systemTable: Table = null): TestingSparkCatalog = { + val catalog = new TestingSparkCatalog(table, systemTable) + 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, systemTable: Table) extends SparkCatalog { + + var lastPaimonIdentifier: Identifier = _ + + override protected def loadSparkTable( + ident: Identifier, + 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 { + + 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 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 = new TestingReadWriteTable + + 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/paimon-spark/pom.xml b/paimon-spark/pom.xml index 7e38786043c7..3be537f55578 100644 --- a/paimon-spark/pom.xml +++ b/paimon-spark/pom.xml @@ -385,4 +385,43 @@ 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 + + + + + integration-tests + + + + **/FlussCatalogReadWriteITCase.* + + + + + + + + + diff --git a/pom.xml b/pom.xml index 5a6bf1a91640..92555a3c8398 100644 --- a/pom.xml +++ b/pom.xml @@ -113,6 +113,9 @@ under the License. true 1.19.1 1.6.1 + 1.0-SNAPSHOT + 2.12 + 5.4.0 1.16.0 1.9.8 3.19.6