entry : options.entrySet()) {
+ if (entry.getKey().startsWith(prefix)) {
+ result.put(entry.getKey().substring(prefix.length()), entry.getValue());
+ }
+ }
+ return result;
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lakestream/FakeStoreCatalog.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lakestream/FakeStoreCatalog.java
new file mode 100644
index 000000000000..8a9f229e74cf
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lakestream/FakeStoreCatalog.java
@@ -0,0 +1,362 @@
+/*
+ * 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.flink.lakestream;
+
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogDatabase;
+import org.apache.flink.table.catalog.CatalogFunction;
+import org.apache.flink.table.catalog.CatalogPartition;
+import org.apache.flink.table.catalog.CatalogPartitionSpec;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
+import org.apache.flink.table.catalog.exceptions.FunctionNotExistException;
+import org.apache.flink.table.catalog.exceptions.PartitionNotExistException;
+import org.apache.flink.table.catalog.exceptions.TableNotExistException;
+import org.apache.flink.table.catalog.stats.CatalogColumnStatistics;
+import org.apache.flink.table.catalog.stats.CatalogTableStatistics;
+import org.apache.flink.table.expressions.Expression;
+import org.apache.flink.table.factories.Factory;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+/**
+ * A fake lake-stream store {@link org.apache.flink.table.catalog.Catalog} for tests. It records
+ * {@link #getTable(ObjectPath)} calls and lifecycle state so tests can verify how {@link
+ * FlinkCatalog} discovers, opens, delegates to and closes the store catalog.
+ *
+ * {@link #getFactory()} exposes the fake store's table factories; the catalog option {@code
+ * factory.mode} ({@code both} (default), {@code source-only}, {@code sink-only}, {@code none})
+ * controls which kind is exposed so Paimon's factory-type validation can be tested.
+ */
+public class FakeStoreCatalog implements org.apache.flink.table.catalog.Catalog {
+
+ public static final String SENTINEL_COMMENT = "fake store table";
+
+ public static final String FACTORY_MODE_OPTION = "factory.mode";
+
+ private static final List GET_TABLE_CALLS = new ArrayList<>();
+
+ /** Paths {@link #getTable(ObjectPath)} reports as absent, so the error paths are reachable. */
+ private static final List MISSING_TABLES =
+ Collections.synchronizedList(new ArrayList());
+
+ private final String name;
+ private final Map options;
+
+ private boolean opened;
+ private boolean closed;
+
+ public FakeStoreCatalog(String name, Map options) {
+ this.name = name;
+ this.options = options;
+ }
+
+ public static List getTableCalls() {
+ return GET_TABLE_CALLS;
+ }
+
+ public static void resetGetTableCalls() {
+ GET_TABLE_CALLS.clear();
+ MISSING_TABLES.clear();
+ }
+
+ public static void markTableMissing(ObjectPath tablePath) {
+ MISSING_TABLES.add(tablePath);
+ }
+
+ public boolean isOpened() {
+ return opened;
+ }
+
+ public boolean isClosed() {
+ return closed;
+ }
+
+ @Override
+ public void open() {
+ opened = true;
+ }
+
+ @Override
+ public void close() {
+ closed = true;
+ }
+
+ @Override
+ public Optional getFactory() {
+ String mode = options.get(FACTORY_MODE_OPTION);
+ if (mode == null || "both".equals(mode)) {
+ return Optional.of(new FakeStoreTableFactory());
+ } else if ("source-only".equals(mode)) {
+ return Optional.of(new FakeStoreTableFactory.FakeStoreSourceFactory());
+ } else if ("sink-only".equals(mode)) {
+ return Optional.of(new FakeStoreTableFactory.FakeStoreSinkFactory());
+ } else if ("none".equals(mode)) {
+ return Optional.empty();
+ }
+ throw new IllegalArgumentException("Unknown " + FACTORY_MODE_OPTION + ": " + mode);
+ }
+
+ @Override
+ public String getDefaultDatabase() {
+ return "default";
+ }
+
+ @Override
+ public List listDatabases() {
+ return Collections.singletonList("default");
+ }
+
+ @Override
+ public CatalogDatabase getDatabase(String databaseName) throws DatabaseNotExistException {
+ throw new DatabaseNotExistException(name, databaseName);
+ }
+
+ @Override
+ public boolean databaseExists(String databaseName) {
+ return "default".equals(databaseName);
+ }
+
+ @Override
+ public void createDatabase(
+ String databaseName, CatalogDatabase database, boolean ignoreIfExists) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void dropDatabase(String databaseName, boolean ignoreIfNotExists, boolean cascade) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void alterDatabase(
+ String databaseName, CatalogDatabase newDatabase, boolean ignoreIfNotExists) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public List listTables(String databaseName) throws DatabaseNotExistException {
+ if (!databaseExists(databaseName)) {
+ throw new DatabaseNotExistException(name, databaseName);
+ }
+ return Collections.emptyList();
+ }
+
+ @Override
+ public List listViews(String databaseName) throws DatabaseNotExistException {
+ if (!databaseExists(databaseName)) {
+ throw new DatabaseNotExistException(name, databaseName);
+ }
+ return Collections.emptyList();
+ }
+
+ @Override
+ public CatalogBaseTable getTable(ObjectPath tablePath) throws TableNotExistException {
+ GET_TABLE_CALLS.add(tablePath);
+ if (MISSING_TABLES.contains(tablePath)) {
+ throw new TableNotExistException(name, tablePath);
+ }
+ Map tableOptions = new HashMap<>();
+ tableOptions.put("connector", FakeStoreCatalogFactory.IDENTIFIER);
+ // CatalogTable#of was removed in Flink 2.x, so build it the way both versions accept
+ return CatalogTable.newBuilder()
+ .schema(Schema.newBuilder().column("id", DataTypes.INT()).build())
+ .comment(SENTINEL_COMMENT)
+ .partitionKeys(Collections.emptyList())
+ .options(tableOptions)
+ .build();
+ }
+
+ @Override
+ public boolean tableExists(ObjectPath tablePath) {
+ return false;
+ }
+
+ @Override
+ public void dropTable(ObjectPath tablePath, boolean ignoreIfNotExists) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignoreIfNotExists) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void alterTable(
+ ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public List listPartitions(ObjectPath tablePath) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public List listPartitions(
+ ObjectPath tablePath, CatalogPartitionSpec partitionSpec) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public List listPartitionsByFilter(
+ ObjectPath tablePath, List filters) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public CatalogPartition getPartition(ObjectPath tablePath, CatalogPartitionSpec partitionSpec)
+ throws PartitionNotExistException {
+ throw new PartitionNotExistException(name, tablePath, partitionSpec);
+ }
+
+ @Override
+ public boolean partitionExists(ObjectPath tablePath, CatalogPartitionSpec partitionSpec) {
+ return false;
+ }
+
+ @Override
+ public void createPartition(
+ ObjectPath tablePath,
+ CatalogPartitionSpec partitionSpec,
+ CatalogPartition partition,
+ boolean ignoreIfExists) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void dropPartition(
+ ObjectPath tablePath, CatalogPartitionSpec partitionSpec, boolean ignoreIfNotExists) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void alterPartition(
+ ObjectPath tablePath,
+ CatalogPartitionSpec partitionSpec,
+ CatalogPartition newPartition,
+ boolean ignoreIfNotExists) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public List listFunctions(String databaseName) throws DatabaseNotExistException {
+ if (!databaseExists(databaseName)) {
+ throw new DatabaseNotExistException(name, databaseName);
+ }
+ return Collections.emptyList();
+ }
+
+ @Override
+ public CatalogFunction getFunction(ObjectPath functionPath) throws FunctionNotExistException {
+ throw new FunctionNotExistException(name, functionPath);
+ }
+
+ @Override
+ public boolean functionExists(ObjectPath functionPath) {
+ return false;
+ }
+
+ @Override
+ public void createFunction(
+ ObjectPath functionPath, CatalogFunction function, boolean ignoreIfExists) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void alterFunction(
+ ObjectPath functionPath, CatalogFunction newFunction, boolean ignoreIfNotExists) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void dropFunction(ObjectPath functionPath, boolean ignoreIfNotExists) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public CatalogTableStatistics getTableStatistics(ObjectPath tablePath) {
+ return CatalogTableStatistics.UNKNOWN;
+ }
+
+ @Override
+ public CatalogColumnStatistics getTableColumnStatistics(ObjectPath tablePath) {
+ return CatalogColumnStatistics.UNKNOWN;
+ }
+
+ @Override
+ public CatalogTableStatistics getPartitionStatistics(
+ ObjectPath tablePath, CatalogPartitionSpec partitionSpec) {
+ return CatalogTableStatistics.UNKNOWN;
+ }
+
+ @Override
+ public CatalogColumnStatistics getPartitionColumnStatistics(
+ ObjectPath tablePath, CatalogPartitionSpec partitionSpec) {
+ return CatalogColumnStatistics.UNKNOWN;
+ }
+
+ @Override
+ public void alterTableStatistics(
+ ObjectPath tablePath,
+ CatalogTableStatistics tableStatistics,
+ boolean ignoreIfNotExists) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void alterTableColumnStatistics(
+ ObjectPath tablePath,
+ CatalogColumnStatistics columnStatistics,
+ boolean ignoreIfNotExists) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void alterPartitionStatistics(
+ ObjectPath tablePath,
+ CatalogPartitionSpec partitionSpec,
+ CatalogTableStatistics partitionStatistics,
+ boolean ignoreIfNotExists) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void alterPartitionColumnStatistics(
+ ObjectPath tablePath,
+ CatalogPartitionSpec partitionSpec,
+ CatalogColumnStatistics columnStatistics,
+ boolean ignoreIfNotExists) {
+ throw new UnsupportedOperationException();
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lakestream/FakeStoreCatalogFactory.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lakestream/FakeStoreCatalogFactory.java
new file mode 100644
index 000000000000..c38c52515b9d
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lakestream/FakeStoreCatalogFactory.java
@@ -0,0 +1,103 @@
+/*
+ * 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.flink.lakestream;
+
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.factories.CatalogFactory;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * A fake lake-stream store {@link CatalogFactory} for tests. It is registered under the identifier
+ * {@code fakestore} in the test-scope {@code
+ * META-INF/services/org.apache.flink.table.factories.Factory} and plays the role that Fluss plays
+ * in production, without any external dependency.
+ */
+public class FakeStoreCatalogFactory implements CatalogFactory {
+
+ public static final String IDENTIFIER = "fakestore";
+
+ public static final ConfigOption BOOTSTRAP_SERVERS =
+ ConfigOptions.key("bootstrap.servers").stringType().noDefaultValue();
+
+ private static final List> CREATED_CATALOG_OPTIONS = new ArrayList<>();
+ private static final List CREATED_CATALOG_CONFIGURATIONS = new ArrayList<>();
+ private static final List CREATED_CATALOGS = new ArrayList<>();
+
+ public static List> createdCatalogOptions() {
+ return CREATED_CATALOG_OPTIONS;
+ }
+
+ /** The Flink configurations the created store catalogs were handed, in creation order. */
+ public static List createdCatalogConfigurations() {
+ return CREATED_CATALOG_CONFIGURATIONS;
+ }
+
+ public static List createdCatalogs() {
+ return CREATED_CATALOGS;
+ }
+
+ public static void reset() {
+ CREATED_CATALOG_OPTIONS.clear();
+ CREATED_CATALOG_CONFIGURATIONS.clear();
+ CREATED_CATALOGS.clear();
+ FakeStoreCatalog.resetGetTableCalls();
+ FakeStoreTableFactory.reset();
+ }
+
+ @Override
+ public String factoryIdentifier() {
+ return IDENTIFIER;
+ }
+
+ @Override
+ public Set> requiredOptions() {
+ return Collections.singleton(BOOTSTRAP_SERVERS);
+ }
+
+ @Override
+ public Set> optionalOptions() {
+ return Collections.emptySet();
+ }
+
+ @Override
+ public Catalog createCatalog(Context context) {
+ Map options = new HashMap<>(context.getOptions());
+ CREATED_CATALOG_OPTIONS.add(options);
+ CREATED_CATALOG_CONFIGURATIONS.add(context.getConfiguration());
+ if (!options.containsKey(BOOTSTRAP_SERVERS.key())) {
+ throw new ValidationException(
+ String.format(
+ "Missing required option '%s' for the '%s' catalog.",
+ BOOTSTRAP_SERVERS.key(), IDENTIFIER));
+ }
+ FakeStoreCatalog catalog = new FakeStoreCatalog(context.getName(), options);
+ CREATED_CATALOGS.add(catalog);
+ return catalog;
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lakestream/FakeStoreData.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lakestream/FakeStoreData.java
new file mode 100644
index 000000000000..be2ded82f573
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lakestream/FakeStoreData.java
@@ -0,0 +1,65 @@
+/*
+ * 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.flink.lakestream;
+
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * The in-memory "storage" of the fake lake-stream store. Rows written through {@link FakeStoreSink}
+ * are appended here per store table (keyed by {@code database.table}), and {@link FakeStoreSource}
+ * reads them back, so tests can assert end-to-end routing without any external store.
+ *
+ * The fake store schema is fixed to a single {@code INT} column, so only single-int rows are
+ * supported.
+ */
+public class FakeStoreData {
+
+ private static final Map> ROWS = new HashMap<>();
+
+ private FakeStoreData() {}
+
+ public static synchronized void add(String tableKey, RowData row) {
+ // copy defensively: the runtime may reuse row instances between records
+ GenericRowData copy = new GenericRowData(row.getArity());
+ for (int i = 0; i < row.getArity(); i++) {
+ copy.setField(i, row.getInt(i));
+ }
+ List rows = ROWS.get(tableKey);
+ if (rows == null) {
+ rows = new ArrayList<>();
+ ROWS.put(tableKey, rows);
+ }
+ rows.add(copy);
+ }
+
+ public static synchronized List rows(String tableKey) {
+ List rows = ROWS.get(tableKey);
+ return rows == null ? new ArrayList() : new ArrayList<>(rows);
+ }
+
+ public static synchronized void reset() {
+ ROWS.clear();
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lakestream/FakeStoreSink.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lakestream/FakeStoreSink.java
new file mode 100644
index 000000000000..95fae331b01d
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lakestream/FakeStoreSink.java
@@ -0,0 +1,186 @@
+/*
+ * 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.flink.lakestream;
+
+import org.apache.flink.api.connector.sink2.InitContext;
+import org.apache.flink.api.connector.sink2.Sink;
+import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.flink.api.connector.sink2.WriterInitContext;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.RowLevelModificationScanContext;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.sink.SinkV2Provider;
+import org.apache.flink.table.connector.sink.abilities.SupportsDeletePushDown;
+import org.apache.flink.table.connector.sink.abilities.SupportsPartitioning;
+import org.apache.flink.table.connector.sink.abilities.SupportsRowLevelDelete;
+import org.apache.flink.table.connector.sink.abilities.SupportsRowLevelDelete.RowLevelDeleteInfo;
+import org.apache.flink.table.connector.sink.abilities.SupportsRowLevelUpdate;
+import org.apache.flink.table.connector.sink.abilities.SupportsRowLevelUpdate.RowLevelUpdateInfo;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.expressions.ResolvedExpression;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+/**
+ * A runnable fake lake-stream store {@link DynamicTableSink}. Rows written through it are appended
+ * to {@link FakeStoreData}; ability callbacks ({@code applyStaticPartition} and the row-level ones)
+ * are recorded statically so tests can assert what the planner applied to the store's sink.
+ *
+ * Deliberately not a {@code SupportsOverwrite}: a real store's sink does not implement it
+ * either, which is what makes an {@code INSERT OVERWRITE} on a lake-stream table fail.
+ */
+public class FakeStoreSink
+ implements DynamicTableSink,
+ SupportsPartitioning,
+ SupportsRowLevelDelete,
+ SupportsRowLevelUpdate,
+ SupportsDeletePushDown {
+
+ private static final List ABILITY_CALLS =
+ Collections.synchronizedList(new ArrayList());
+
+ private final String tableKey;
+
+ public FakeStoreSink(String tableKey) {
+ this.tableKey = tableKey;
+ }
+
+ /** Recorded ability callbacks, e.g. {@code applyStaticPartition} / {@code executeDeletion}. */
+ public static List abilityCalls() {
+ synchronized (ABILITY_CALLS) {
+ return new ArrayList<>(ABILITY_CALLS);
+ }
+ }
+
+ public static void resetAbilityCalls() {
+ ABILITY_CALLS.clear();
+ }
+
+ public String tableKey() {
+ return tableKey;
+ }
+
+ @Override
+ public ChangelogMode getChangelogMode(ChangelogMode requestedMode) {
+ return requestedMode;
+ }
+
+ @Override
+ public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
+ return SinkV2Provider.of(new FakeStoreWriteSink(tableKey));
+ }
+
+ @Override
+ public DynamicTableSink copy() {
+ return new FakeStoreSink(tableKey);
+ }
+
+ @Override
+ public String asSummaryString() {
+ return "FakeStoreSink";
+ }
+
+ @Override
+ public void applyStaticPartition(Map staticPartition) {
+ ABILITY_CALLS.add("applyStaticPartition(" + staticPartition + ")");
+ }
+
+ @Override
+ public RowLevelDeleteInfo applyRowLevelDelete(
+ @Nullable RowLevelModificationScanContext context) {
+ ABILITY_CALLS.add("applyRowLevelDelete");
+ return new RowLevelDeleteInfo() {};
+ }
+
+ @Override
+ public RowLevelUpdateInfo applyRowLevelUpdate(
+ List updatedColumns, @Nullable RowLevelModificationScanContext context) {
+ ABILITY_CALLS.add("applyRowLevelUpdate(" + updatedColumns.size() + ")");
+ return new RowLevelUpdateInfo() {};
+ }
+
+ @Override
+ public boolean applyDeleteFilters(List filters) {
+ ABILITY_CALLS.add("applyDeleteFilters(" + filters.size() + ")");
+ return false;
+ }
+
+ @Override
+ public Optional executeDeletion() {
+ ABILITY_CALLS.add("executeDeletion");
+ return Optional.empty();
+ }
+
+ /** The sink2 runtime sink that appends to {@link FakeStoreData}. */
+ static class FakeStoreWriteSink implements Sink {
+
+ private static final long serialVersionUID = 1L;
+
+ private final String tableKey;
+
+ FakeStoreWriteSink(String tableKey) {
+ this.tableKey = tableKey;
+ }
+
+ /**
+ * Do not annotate with @override here to maintain compatibility with Flink
+ * 2.0+.
+ */
+ public SinkWriter createWriter(InitContext context) {
+ return new SinkWriter() {
+ @Override
+ public void write(RowData element, Context context) {
+ FakeStoreData.add(tableKey, element);
+ }
+
+ @Override
+ public void flush(boolean endOfInput) {}
+
+ @Override
+ public void close() {}
+ };
+ }
+
+ /**
+ * Do not annotate with @override here to maintain compatibility with Flink
+ * 1.18-.
+ */
+ public SinkWriter createWriter(WriterInitContext context) {
+ return new SinkWriter() {
+ @Override
+ public void write(RowData element, Context context) {
+ FakeStoreData.add(tableKey, element);
+ }
+
+ @Override
+ public void flush(boolean endOfInput) {}
+
+ @Override
+ public void close() {}
+ };
+ }
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lakestream/FakeStoreSource.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lakestream/FakeStoreSource.java
new file mode 100644
index 000000000000..1ee1d181ca2c
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lakestream/FakeStoreSource.java
@@ -0,0 +1,83 @@
+/*
+ * 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.flink.lakestream;
+
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.ProviderContext;
+import org.apache.flink.table.connector.source.DataStreamScanProvider;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.connector.source.ScanTableSource;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.apache.flink.table.types.logical.RowType;
+
+/**
+ * A runnable fake lake-stream store {@link ScanTableSource}. It emits the rows that were written to
+ * {@link FakeStoreData} for its table key, standing in for the store's union read of un-tiered
+ * data.
+ */
+public class FakeStoreSource implements ScanTableSource {
+
+ private final String tableKey;
+ private final RowType rowType;
+
+ public FakeStoreSource(String tableKey, RowType rowType) {
+ this.tableKey = tableKey;
+ this.rowType = rowType;
+ }
+
+ /** The store table key this source reads, {@code database.table}. */
+ public String tableKey() {
+ return tableKey;
+ }
+
+ @Override
+ public ChangelogMode getChangelogMode() {
+ return ChangelogMode.insertOnly();
+ }
+
+ @Override
+ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext runtimeProviderContext) {
+ return new DataStreamScanProvider() {
+ @Override
+ public DataStream produceDataStream(
+ ProviderContext providerContext, StreamExecutionEnvironment executor) {
+ return executor.fromData(
+ FakeStoreData.rows(tableKey), InternalTypeInfo.of(rowType));
+ }
+
+ @Override
+ public boolean isBounded() {
+ return true;
+ }
+ };
+ }
+
+ @Override
+ public DynamicTableSource copy() {
+ return new FakeStoreSource(tableKey, rowType);
+ }
+
+ @Override
+ public String asSummaryString() {
+ return "FakeStoreSource";
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lakestream/FakeStoreTableFactory.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lakestream/FakeStoreTableFactory.java
new file mode 100644
index 000000000000..dc2d24ef0626
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lakestream/FakeStoreTableFactory.java
@@ -0,0 +1,169 @@
+/*
+ * 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.flink.lakestream;
+
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.factories.DynamicTableFactory;
+import org.apache.flink.table.factories.DynamicTableSinkFactory;
+import org.apache.flink.table.factories.DynamicTableSourceFactory;
+import org.apache.flink.table.types.logical.RowType;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * The fake lake-stream store's table factory, playing the role that a store's catalog-scoped
+ * factory (e.g. Fluss's) plays in production. {@link FakeStoreCatalog#getFactory()} returns it, and
+ * Paimon's lake-stream routing delegates {@code $rt} source creation and regular sink creation
+ * through it.
+ *
+ * The contexts it is invoked with are recorded statically so tests can assert the adaptation
+ * Paimon performs (base identifier without {@code $rt}, store table options, prefix-removed store
+ * connection options).
+ */
+public class FakeStoreTableFactory implements DynamicTableSourceFactory, DynamicTableSinkFactory {
+
+ public static final String IDENTIFIER = FakeStoreCatalogFactory.IDENTIFIER;
+
+ private static final List SOURCE_CONTEXTS =
+ Collections.synchronizedList(new ArrayList());
+ private static final List SINK_CONTEXTS =
+ Collections.synchronizedList(new ArrayList());
+
+ private final FakeStoreSourceFactory sourceFactory = new FakeStoreSourceFactory();
+ private final FakeStoreSinkFactory sinkFactory = new FakeStoreSinkFactory();
+
+ public static List sourceContexts() {
+ synchronized (SOURCE_CONTEXTS) {
+ return new ArrayList<>(SOURCE_CONTEXTS);
+ }
+ }
+
+ public static List sinkContexts() {
+ synchronized (SINK_CONTEXTS) {
+ return new ArrayList<>(SINK_CONTEXTS);
+ }
+ }
+
+ public static void reset() {
+ SOURCE_CONTEXTS.clear();
+ SINK_CONTEXTS.clear();
+ FakeStoreSink.resetAbilityCalls();
+ FakeStoreData.reset();
+ }
+
+ static void recordSourceContext(DynamicTableFactory.Context context) {
+ SOURCE_CONTEXTS.add(context);
+ }
+
+ static void recordSinkContext(DynamicTableFactory.Context context) {
+ SINK_CONTEXTS.add(context);
+ }
+
+ static String tableKeyOf(DynamicTableFactory.Context context) {
+ return context.getObjectIdentifier().getDatabaseName()
+ + "."
+ + context.getObjectIdentifier().getObjectName();
+ }
+
+ @Override
+ public String factoryIdentifier() {
+ return IDENTIFIER;
+ }
+
+ @Override
+ public Set> requiredOptions() {
+ return Collections.emptySet();
+ }
+
+ @Override
+ public Set> optionalOptions() {
+ return Collections.emptySet();
+ }
+
+ @Override
+ public DynamicTableSource createDynamicTableSource(Context context) {
+ return sourceFactory.createDynamicTableSource(context);
+ }
+
+ @Override
+ public DynamicTableSink createDynamicTableSink(Context context) {
+ return sinkFactory.createDynamicTableSink(context);
+ }
+
+ /** Source-only variant, to test Paimon's factory-type validation. */
+ static class FakeStoreSourceFactory implements DynamicTableSourceFactory {
+
+ @Override
+ public String factoryIdentifier() {
+ return IDENTIFIER;
+ }
+
+ @Override
+ public Set> requiredOptions() {
+ return Collections.emptySet();
+ }
+
+ @Override
+ public Set> optionalOptions() {
+ return Collections.emptySet();
+ }
+
+ @Override
+ public DynamicTableSource createDynamicTableSource(Context context) {
+ recordSourceContext(context);
+ return new FakeStoreSource(
+ tableKeyOf(context),
+ (RowType)
+ context.getCatalogTable()
+ .getResolvedSchema()
+ .toPhysicalRowDataType()
+ .getLogicalType());
+ }
+ }
+
+ /** Sink-only variant, to test Paimon's factory-type validation. */
+ static class FakeStoreSinkFactory implements DynamicTableSinkFactory {
+
+ @Override
+ public String factoryIdentifier() {
+ return IDENTIFIER;
+ }
+
+ @Override
+ public Set> requiredOptions() {
+ return Collections.emptySet();
+ }
+
+ @Override
+ public Set> optionalOptions() {
+ return Collections.emptySet();
+ }
+
+ @Override
+ public DynamicTableSink createDynamicTableSink(Context context) {
+ recordSinkContext(context);
+ return new FakeStoreSink(tableKeyOf(context));
+ }
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lakestream/LakeStreamCatalogTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lakestream/LakeStreamCatalogTest.java
new file mode 100644
index 000000000000..6910238409a5
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lakestream/LakeStreamCatalogTest.java
@@ -0,0 +1,514 @@
+/*
+ * 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.flink.lakestream;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.catalog.CatalogFactory;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.flink.DataCatalogTable;
+import org.apache.paimon.flink.FlinkCatalog;
+import org.apache.paimon.flink.FlinkCatalogFactory;
+import org.apache.paimon.flink.FlinkCatalogOptions;
+import org.apache.paimon.flink.FlinkTableFactory;
+import org.apache.paimon.flink.SystemCatalogTable;
+import org.apache.paimon.options.CatalogOptions;
+import org.apache.paimon.options.Options;
+
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogDatabaseImpl;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.ResolvedCatalogTable;
+import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.catalog.WatermarkSpec;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.TableNotExistException;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for the lake-stream wiring in {@link FlinkCatalog}. */
+public class LakeStreamCatalogTest {
+
+ private static final ObjectPath T1 = new ObjectPath("db1", "t1");
+ private static final ObjectPath T1_RT = new ObjectPath("db1", "t1$rt");
+
+ @TempDir java.nio.file.Path tempDir;
+
+ @BeforeEach
+ public void beforeEach() {
+ FakeStoreCatalogFactory.reset();
+ }
+
+ // ------------------------------------------------------------------------
+ // helpers
+ // ------------------------------------------------------------------------
+
+ private Options conf(String storeIdentifier) {
+ Options conf = new Options();
+ conf.setString(CatalogOptions.WAREHOUSE.key(), tempDir.toString());
+ if (storeIdentifier != null) {
+ conf.setString(CatalogOptions.LAKE_STREAM_STORE.key(), storeIdentifier);
+ }
+ return conf;
+ }
+
+ private FlinkCatalog newCatalog(Options conf) {
+ CatalogContext context = CatalogContext.create(conf);
+ return new FlinkCatalog(
+ CatalogFactory.createCatalog(context, getClass().getClassLoader()),
+ "test-catalog",
+ context.options().get(FlinkCatalogOptions.DEFAULT_DATABASE),
+ context.options(),
+ new Configuration());
+ }
+
+ private FlinkCatalog catalogWithStore() {
+ Options conf = conf(FakeStoreCatalogFactory.IDENTIFIER);
+ conf.setString("fakestore.bootstrap.servers", "localhost:9123");
+ return newCatalog(conf);
+ }
+
+ private void createTable(FlinkCatalog catalog, ObjectPath path, boolean lakeStreamEnabled)
+ throws Exception {
+ catalog.createDatabase(
+ path.getDatabaseName(), new CatalogDatabaseImpl(new HashMap<>(), null), true);
+ Map options = new HashMap<>();
+ options.put(CoreOptions.BUCKET.key(), "1");
+ options.put(CoreOptions.PRIMARY_KEY.key(), "id");
+ if (lakeStreamEnabled) {
+ options.put(CoreOptions.LAKE_STREAM_ENABLED.key(), "true");
+ }
+ ResolvedSchema resolvedSchema =
+ new ResolvedSchema(
+ Collections.singletonList(Column.physical("id", DataTypes.INT().notNull())),
+ Collections.emptyList(),
+ null);
+ CatalogTable table =
+ new ResolvedCatalogTable(
+ CatalogTable.newBuilder()
+ .schema(
+ Schema.newBuilder()
+ .fromResolvedSchema(resolvedSchema)
+ .build())
+ .partitionKeys(Collections.emptyList())
+ .options(options)
+ .build(),
+ resolvedSchema);
+ catalog.createTable(path, table, false);
+ }
+
+ // ------------------------------------------------------------------------
+ // option
+ // ------------------------------------------------------------------------
+
+ @Test
+ public void testLakeStreamStoreOptionDeclared() {
+ assertThat(CatalogOptions.LAKE_STREAM_STORE.key()).isEqualTo("lakestream.store");
+ assertThat(CatalogOptions.LAKE_STREAM_STORE.hasDefaultValue()).isFalse();
+ }
+
+ // ------------------------------------------------------------------------
+ // store catalog discovery (PIP-39 L209-250)
+ // ------------------------------------------------------------------------
+
+ @Test
+ public void testGetOrCreateStoreCatalogExtractsPrefixedOptions() throws Exception {
+ Options conf = conf(FakeStoreCatalogFactory.IDENTIFIER);
+ conf.setString("fakestore.bootstrap.servers", "localhost:9123");
+ conf.setString("fakestore.extra", "abc");
+ conf.setString("some.unrelated", "xyz");
+ FlinkCatalog catalog = newCatalog(conf);
+
+ org.apache.flink.table.catalog.Catalog store = catalog.getOrCreateLakeStreamStoreCatalog();
+
+ assertThat(store).isInstanceOf(FakeStoreCatalog.class);
+ assertThat(((FakeStoreCatalog) store).isOpened()).isTrue();
+ assertThat(FakeStoreCatalogFactory.createdCatalogOptions()).hasSize(1);
+ Map passed = FakeStoreCatalogFactory.createdCatalogOptions().get(0);
+ assertThat(passed)
+ .containsEntry("bootstrap.servers", "localhost:9123")
+ .containsEntry("extra", "abc");
+ assertThat(passed)
+ .doesNotContainKeys(
+ "fakestore.bootstrap.servers",
+ "fakestore.extra",
+ CatalogOptions.LAKE_STREAM_STORE.key(),
+ CatalogOptions.WAREHOUSE.key(),
+ "some.unrelated");
+ catalog.close();
+ }
+
+ @Test
+ public void testStoreCatalogReceivesCatalogConfigurationWhenFirstCreatedViaGetTable()
+ throws Exception {
+ Options conf = conf(FakeStoreCatalogFactory.IDENTIFIER);
+ conf.setString("fakestore.bootstrap.servers", "localhost:9123");
+
+ // stands in for a platform-injected credential carried by the Flink configuration of the
+ // CREATE CATALOG statement rather than by a catalog option
+ ConfigOption tokenPath =
+ ConfigOptions.key("platform.injected.token.path").stringType().noDefaultValue();
+ Configuration flinkConf = new Configuration();
+ flinkConf.set(tokenPath, "/var/run/secrets/token");
+
+ FlinkCatalog catalog =
+ new FlinkCatalogFactory()
+ .createCatalog(
+ new FactoryUtil.DefaultCatalogContext(
+ "test-catalog",
+ conf.toMap(),
+ flinkConf,
+ getClass().getClassLoader()));
+ try {
+ createTable(catalog, T1, true);
+
+ // the planner resolves table metadata before it creates a source, so the store catalog
+ // is normally first created through getTable(), which has no table-factory context
+ catalog.getTable(T1_RT);
+
+ assertThat(FakeStoreCatalogFactory.createdCatalogConfigurations()).hasSize(1);
+ assertThat(FakeStoreCatalogFactory.createdCatalogConfigurations().get(0).get(tokenPath))
+ .isEqualTo("/var/run/secrets/token");
+ } finally {
+ catalog.close();
+ }
+ }
+
+ @Test
+ public void testStoreCatalogIsCachedAndReused() throws Exception {
+ FlinkCatalog catalog = catalogWithStore();
+
+ org.apache.flink.table.catalog.Catalog first = catalog.getOrCreateLakeStreamStoreCatalog();
+ org.apache.flink.table.catalog.Catalog second = catalog.getOrCreateLakeStreamStoreCatalog();
+
+ assertThat(second).isSameAs(first);
+ assertThat(FakeStoreCatalogFactory.createdCatalogs()).hasSize(1);
+ catalog.close();
+ }
+
+ @Test
+ public void testMissingStoreOptionFails() throws Exception {
+ FlinkCatalog catalog = newCatalog(conf(null));
+
+ assertThatThrownBy(() -> catalog.getOrCreateLakeStreamStoreCatalog())
+ .isInstanceOf(CatalogException.class)
+ .hasMessageContaining(CatalogOptions.LAKE_STREAM_STORE.key());
+ catalog.close();
+ }
+
+ @Test
+ public void testUndiscoverableStoreIdentifierFails() throws Exception {
+ FlinkCatalog catalog = newCatalog(conf("nosuchstore"));
+
+ assertThatThrownBy(() -> catalog.getOrCreateLakeStreamStoreCatalog())
+ .isInstanceOf(CatalogException.class)
+ .hasMessageContaining("nosuchstore");
+ catalog.close();
+ }
+
+ @Test
+ public void testMissingRequiredProviderOptionFails() throws Exception {
+ // fakestore requires 'bootstrap.servers' (as 'fakestore.bootstrap.servers' in the catalog
+ // options); it is deliberately missing here.
+ FlinkCatalog catalog = newCatalog(conf(FakeStoreCatalogFactory.IDENTIFIER));
+
+ assertThatThrownBy(() -> catalog.getOrCreateLakeStreamStoreCatalog())
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("bootstrap.servers");
+ catalog.close();
+ }
+
+ @Test
+ public void testCloseClosesCachedStoreCatalog() throws Exception {
+ FlinkCatalog catalog = catalogWithStore();
+ catalog.getOrCreateLakeStreamStoreCatalog();
+
+ catalog.close();
+
+ assertThat(FakeStoreCatalogFactory.createdCatalogs()).hasSize(1);
+ assertThat(FakeStoreCatalogFactory.createdCatalogs().get(0).isClosed()).isTrue();
+ }
+
+ @Test
+ public void testCloseWithoutStoreCatalogIsHarmless() throws Exception {
+ FlinkCatalog catalog = catalogWithStore();
+
+ assertThatCode(catalog::close).doesNotThrowAnyException();
+ assertThat(FakeStoreCatalogFactory.createdCatalogs()).isEmpty();
+ }
+
+ // ------------------------------------------------------------------------
+ // $rt interception (PIP-39 L98-116, L254)
+ // ------------------------------------------------------------------------
+
+ @Test
+ public void testGetTableRealTimeRoutesToStoreWithBaseIdentifier() throws Exception {
+ FlinkCatalog catalog = catalogWithStore();
+ createTable(catalog, T1, true);
+
+ CatalogBaseTable result = catalog.getTable(T1_RT);
+
+ assertThat(result.getComment()).isEqualTo(FakeStoreCatalog.SENTINEL_COMMENT);
+ assertThat(FakeStoreCatalog.getTableCalls()).containsExactly(T1);
+ catalog.close();
+ }
+
+ @Test
+ public void testGetTableRealTimeThrowsWhenLakeStreamDisabled() throws Exception {
+ FlinkCatalog catalog = catalogWithStore();
+ createTable(catalog, T1, false);
+
+ assertThatThrownBy(() -> catalog.getTable(T1_RT))
+ .isInstanceOf(TableNotExistException.class);
+ assertThat(FakeStoreCatalog.getTableCalls()).isEmpty();
+ catalog.close();
+ }
+
+ @Test
+ public void testGetTableRealTimeThrowsWhenBaseTableMissing() throws Exception {
+ FlinkCatalog catalog = catalogWithStore();
+
+ assertThatThrownBy(() -> catalog.getTable(new ObjectPath("db1", "no_such$rt")))
+ .isInstanceOf(TableNotExistException.class);
+ assertThat(FakeStoreCatalog.getTableCalls()).isEmpty();
+ catalog.close();
+ }
+
+ @Test
+ public void testGetTableRealTimeFailsClearlyWhenTheStoreSideIsMissing() throws Exception {
+ FlinkCatalog catalog = catalogWithStore();
+ createTable(catalog, T1, true);
+ FakeStoreCatalog.markTableMissing(T1);
+
+ // the store's own not-exists error would name the base table and this catalog, which does
+ // not tell the user that the store side is what is absent
+ assertThatThrownBy(() -> catalog.getTable(T1_RT))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("does not exist in the Lakestream Store")
+ .hasMessageContaining(FakeStoreCatalogFactory.IDENTIFIER);
+ catalog.close();
+ }
+
+ @Test
+ public void testGetTableRealTimeNamesTheMissingStoreOption() throws Exception {
+ FlinkCatalog catalog = newCatalog(conf(null));
+ createTable(catalog, T1, true);
+
+ // the '$rt' name is served by the store and this catalog names no store, so the
+ // failure has to say that; reporting an unknown table would send the user looking for a
+ // table that is there
+ assertThatThrownBy(() -> catalog.getTable(T1_RT))
+ .isInstanceOf(CatalogException.class)
+ .hasMessageContaining(CatalogOptions.LAKE_STREAM_STORE.key())
+ .hasMessageContaining("test-catalog");
+ catalog.close();
+ }
+
+ @Test
+ public void testGetTableRealTimeRejectsTimeTravel() throws Exception {
+ FlinkCatalog catalog = catalogWithStore();
+ createTable(catalog, T1, true);
+
+ assertThatThrownBy(() -> catalog.getTable(T1_RT, 12345L))
+ .isInstanceOf(CatalogException.class)
+ .hasMessageContaining("time travel");
+ assertThat(FakeStoreCatalog.getTableCalls()).isEmpty();
+ catalog.close();
+ }
+
+ @Test
+ public void testGetTableTimeTravelDelegatesForNormalTable() throws Exception {
+ FlinkCatalog catalog = catalogWithStore();
+ createTable(catalog, T1, true);
+
+ CatalogBaseTable result = catalog.getTable(T1, 12345L);
+
+ assertThat(result.getOptions())
+ .containsEntry(CoreOptions.SCAN_TIMESTAMP_MILLIS.key(), "12345");
+ assertThat(FakeStoreCatalog.getTableCalls()).isEmpty();
+ catalog.close();
+ }
+
+ @Test
+ public void testBranchSuffixIsNotTreatedAsRealTime() throws Exception {
+ FlinkCatalog catalog = catalogWithStore();
+ createTable(catalog, T1, true);
+
+ assertThatThrownBy(() -> catalog.getTable(new ObjectPath("db1", "t1$branch_b1$rt")))
+ .isInstanceOf(TableNotExistException.class);
+ assertThat(FakeStoreCatalog.getTableCalls()).isEmpty();
+ catalog.close();
+ }
+
+ @Test
+ public void testPaimonSystemTableStillDelegates() throws Exception {
+ FlinkCatalog catalog = catalogWithStore();
+ createTable(catalog, T1, true);
+
+ CatalogBaseTable result = catalog.getTable(new ObjectPath("db1", "t1$snapshots"));
+
+ assertThat(result).isInstanceOf(SystemCatalogTable.class);
+ assertThat(FakeStoreCatalog.getTableCalls()).isEmpty();
+ catalog.close();
+ }
+
+ // ------------------------------------------------------------------------
+ // recovery from a mis-configured table
+ // ------------------------------------------------------------------------
+
+ @Test
+ public void testCorrectingAMisconfiguredLakeStreamTableNeverContactsTheStore()
+ throws Exception {
+ // an identifier that no CatalogFactory answers to, so any attempt to create the store
+ // catalog would fail and make this test fail
+ FlinkCatalog catalog = newCatalog(conf("nonexistent-store"));
+ createTable(catalog, T1, true);
+
+ catalog.dropTable(T1, false);
+ createTable(catalog, T1, false);
+
+ assertThat(FakeStoreCatalogFactory.createdCatalogs()).isEmpty();
+ assertThat(catalog.tableExists(T1)).isTrue();
+ catalog.close();
+ }
+
+ // ------------------------------------------------------------------------
+ // inherited Paimon catalog behaviour
+ // ------------------------------------------------------------------------
+
+ @Test
+ public void testInheritedPaimonCatalogBehaviourIsIntact() throws Exception {
+ FlinkCatalog catalog = catalogWithStore();
+ createTable(catalog, T1, true);
+
+ assertThat(catalog.getName()).isEqualTo("test-catalog");
+ assertThat(catalog.getDefaultDatabase()).isEqualTo("default");
+ assertThat(catalog.listDatabases()).contains("db1", "default");
+ assertThat(catalog.databaseExists("db1")).isTrue();
+ assertThat(catalog.getDatabase("db1")).isNotNull();
+ assertThat(catalog.listTables("db1")).contains("t1");
+ assertThat(catalog.listViews("db1")).isEmpty();
+ assertThat(catalog.tableExists(T1)).isTrue();
+ assertThat(catalog.tableExists(new ObjectPath("db1", "no_such"))).isFalse();
+ assertThat(catalog.getTable(T1)).isInstanceOf(DataCatalogTable.class);
+ assertThat(catalog.getTableStatistics(T1)).isNotNull();
+ assertThat(catalog.listFunctions("db1")).isEmpty();
+ assertThat(catalog.functionExists(new ObjectPath("db1", "f"))).isFalse();
+
+ List procedures = catalog.listProcedures("db1");
+ assertThat(procedures).isNotEmpty();
+ // paimon procedures are only served from the 'sys' database
+ assertThat(catalog.getProcedure(new ObjectPath("sys", procedures.get(0)))).isNotNull();
+
+ assertThat(catalog.getFactory()).isPresent();
+ assertThat(catalog.getFactory().get()).isInstanceOf(FlinkTableFactory.class);
+
+ // it is still a Paimon Flink catalog, so the Paimon catalog behind it stays reachable
+ assertThat(catalog).isInstanceOf(FlinkCatalog.class);
+ assertThat(
+ catalog.catalog()
+ .getTable(
+ Identifier.create(
+ T1.getDatabaseName(), T1.getObjectName())))
+ .isNotNull();
+
+ catalog.dropTable(T1, false);
+ assertThat(catalog.tableExists(T1)).isFalse();
+ catalog.close();
+ }
+
+ // ------------------------------------------------------------------------
+ // table factory plumbing (spec Part 8)
+ // ------------------------------------------------------------------------
+
+ @Test
+ public void testFlinkTableFactoryConstructorOverloads() throws Exception {
+ FlinkCatalog catalog = catalogWithStore();
+
+ assertThat(new FlinkTableFactory().factoryIdentifier())
+ .isEqualTo(FlinkCatalogFactory.IDENTIFIER);
+ assertThat(new FlinkTableFactory(null, null).factoryIdentifier())
+ .isEqualTo(FlinkCatalogFactory.IDENTIFIER);
+ assertThat(
+ new FlinkTableFactory(catalog, new LakeStreamTableFactory(catalog))
+ .factoryIdentifier())
+ .isEqualTo(FlinkCatalogFactory.IDENTIFIER);
+ catalog.close();
+ }
+
+ // ------------------------------------------------------------------------
+ // FlinkCatalogFactory
+ // ------------------------------------------------------------------------
+
+ @Test
+ public void testFlinkCatalogFactoryEnablesLakeStreamWhenStoreConfigured() {
+ Options conf = conf(FakeStoreCatalogFactory.IDENTIFIER);
+ conf.setString("fakestore.bootstrap.servers", "localhost:9123");
+
+ FlinkCatalog catalog =
+ new FlinkCatalogFactory()
+ .createCatalog(
+ new FactoryUtil.DefaultCatalogContext(
+ "test-catalog",
+ conf.toMap(),
+ new Configuration(),
+ getClass().getClassLoader()));
+
+ assertThat(catalog.lakeStreamStoreConfigured()).isTrue();
+ assertThat(catalog.catalog()).isNotNull();
+ catalog.close();
+ }
+
+ @Test
+ public void testFlinkCatalogFactoryLeavesLakeStreamOffWithoutStore() {
+ Options conf = conf(null);
+
+ FlinkCatalog catalog =
+ new FlinkCatalogFactory()
+ .createCatalog(
+ new FactoryUtil.DefaultCatalogContext(
+ "test-catalog",
+ conf.toMap(),
+ new Configuration(),
+ getClass().getClassLoader()));
+
+ assertThat(catalog).isInstanceOf(FlinkCatalog.class);
+ assertThat(catalog.lakeStreamStoreConfigured()).isFalse();
+ catalog.close();
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lakestream/LakeStreamFlussITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lakestream/LakeStreamFlussITCase.java
new file mode 100644
index 000000000000..6822d372c86d
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lakestream/LakeStreamFlussITCase.java
@@ -0,0 +1,180 @@
+/*
+ * 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.flink.lakestream;
+
+import org.apache.paimon.flink.FlinkCatalogFactory;
+import org.apache.paimon.options.CatalogOptions;
+import org.apache.paimon.options.Options;
+
+import org.apache.flink.api.common.RuntimeExecutionMode;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CloseableIterator;
+import org.apache.fluss.server.testutils.FlussClusterExtension;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * Runs the lake-stream routing against a real Fluss cluster that Paimon reaches only through
+ * Flink's {@code CatalogFactory} SPI. Paimon compiles against no Fluss class, so what this test
+ * exercises is what a user gets from a published Fluss release.
+ *
+ * It covers what the fake store in {@link LakeStreamRoutingITCase} cannot: that the discovered
+ * Fluss catalog accepts the options forwarded under the {@code fluss.} prefix, that a missing
+ * required store option surfaces as a configuration error, and that reads of the {@code $rt} name
+ * are served by the Fluss source, query hints included.
+ *
+ *
A Fluss log table is unbounded, so the {@code $rt} reads pin a bounded scan window to keep the
+ * assertions deterministic.
+ */
+public class LakeStreamFlussITCase {
+
+ private static final String STORE_IDENTIFIER = "fluss";
+ private static final String DATABASE = FlussClusterExtension.BUILTIN_DATABASE;
+
+ /** Reads the log as it stands when the query starts, instead of tailing it forever. */
+ private static final String BOUNDED_SCAN =
+ "/*+ OPTIONS('scan.startup.mode' = 'earliest', "
+ + "'scan.bounded.mode' = 'latest-offset') */";
+
+ @RegisterExtension
+ static final FlussClusterExtension FLUSS =
+ FlussClusterExtension.builder().setNumOfTabletServers(1).build();
+
+ @TempDir java.nio.file.Path tempDir;
+
+ private Catalog catalog;
+
+ @BeforeEach
+ public void beforeEach() {
+ catalog = newCatalog(withStoreConnection(baseOptions()));
+ }
+
+ @AfterEach
+ public void afterEach() throws Exception {
+ if (catalog != null) {
+ catalog.close();
+ }
+ }
+
+ private Options baseOptions() {
+ Options conf = new Options();
+ conf.setString(CatalogOptions.WAREHOUSE.key(), tempDir.toString());
+ conf.setString(CatalogOptions.LAKE_STREAM_STORE.key(), STORE_IDENTIFIER);
+ return conf;
+ }
+
+ private Options withStoreConnection(Options conf) {
+ conf.setString(STORE_IDENTIFIER + ".bootstrap.servers", FLUSS.getBootstrapServers());
+ return conf;
+ }
+
+ private Catalog newCatalog(Options conf) {
+ return new FlinkCatalogFactory()
+ .createCatalog(
+ new FactoryUtil.DefaultCatalogContext(
+ "paimon",
+ conf.toMap(),
+ new Configuration(),
+ getClass().getClassLoader()));
+ }
+
+ private StreamTableEnvironment newTableEnv() {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment(1);
+ env.setRuntimeMode(RuntimeExecutionMode.BATCH);
+ StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
+ tEnv.registerCatalog("paimon", catalog);
+ tEnv.executeSql(
+ String.format(
+ "CREATE CATALOG fluss_store WITH ('type' = '%s', 'bootstrap.servers' = '%s')",
+ STORE_IDENTIFIER, FLUSS.getBootstrapServers()));
+ tEnv.useCatalog("paimon");
+ tEnv.executeSql(String.format("CREATE DATABASE IF NOT EXISTS %s", DATABASE));
+ tEnv.useDatabase(DATABASE);
+ return tEnv;
+ }
+
+ /** Creates the store-side table and the matching lake-stream table in Paimon. */
+ private void createLakeStreamTable(StreamTableEnvironment tEnv, String name) {
+ tEnv.executeSql(String.format("CREATE TABLE fluss_store.%s.%s (id INT)", DATABASE, name));
+ tEnv.executeSql(
+ String.format(
+ "CREATE TABLE %s (id INT) WITH ('lakestream.enabled' = 'true')", name));
+ }
+
+ private static List query(StreamTableEnvironment tEnv, String sql) throws Exception {
+ List result = new ArrayList<>();
+ try (CloseableIterator it = tEnv.executeSql(sql).collect()) {
+ while (it.hasNext()) {
+ result.add((Integer) it.next().getField(0));
+ }
+ }
+ return result;
+ }
+
+ @Test
+ public void testRegularWriteGoesToFlussAndTheRealTimeNameReadsItBack() throws Exception {
+ StreamTableEnvironment tEnv = newTableEnv();
+ createLakeStreamTable(tEnv, "orders");
+
+ tEnv.executeSql("INSERT INTO orders SELECT * FROM (VALUES (1), (2)) AS t(id)").await();
+
+ // the regular write reached Fluss, and the '$rt' name reads it back from the store
+ assertThat(query(tEnv, "SELECT id FROM orders$rt " + BOUNDED_SCAN))
+ .containsExactlyInAnyOrder(1, 2);
+
+ // the original name still reads Paimon only, which the regular write bypassed
+ assertThat(query(tEnv, "SELECT id FROM orders")).isEmpty();
+ }
+
+ @Test
+ public void testRealTimeReadInBatchModeNeedsABoundedScan() {
+ StreamTableEnvironment tEnv = newTableEnv();
+ createLakeStreamTable(tEnv, "orders");
+
+ // the store serves '$rt' as an unbounded source, which a batch query cannot read
+ assertThatThrownBy(() -> query(tEnv, "SELECT id FROM orders$rt"))
+ .hasStackTraceContaining("unbounded");
+ }
+
+ @Test
+ public void testMissingStoreOptionFailsWithConfigurationError() {
+ catalog = newCatalog(baseOptions());
+ StreamTableEnvironment tEnv = newTableEnv();
+ tEnv.executeSql("CREATE TABLE broken (id INT) WITH ('lakestream.enabled' = 'true')");
+
+ // the Fluss catalog factory requires 'bootstrap.servers', which was never forwarded
+ assertThatThrownBy(() -> query(tEnv, "SELECT id FROM broken$rt " + BOUNDED_SCAN))
+ .hasStackTraceContaining(STORE_IDENTIFIER)
+ .hasStackTraceContaining("bootstrap.servers");
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lakestream/LakeStreamRoutingITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lakestream/LakeStreamRoutingITCase.java
new file mode 100644
index 000000000000..3b125cb0f060
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lakestream/LakeStreamRoutingITCase.java
@@ -0,0 +1,256 @@
+/*
+ * 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.flink.lakestream;
+
+import org.apache.paimon.flink.FlinkCatalogFactory;
+import org.apache.paimon.options.CatalogOptions;
+import org.apache.paimon.options.Options;
+
+import org.apache.flink.api.common.RuntimeExecutionMode;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.factories.DynamicTableFactory;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CloseableIterator;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * End-to-end tests for the Flink half of the PIP-39 test plan (L506-512), using the fake
+ * lake-stream store:
+ *
+ *
+ * - regular writes on a lake-stream table are routed to the store sink;
+ *
- {@code INSERT OVERWRITE} is rejected, because the store's sink cannot execute it;
+ *
- reads of the original table name access Paimon only;
+ *
- reads of the {@code $rt} name are routed to the store with the BASE identifier;
+ *
- original-name reads succeed with no store configured;
+ *
- non-lake-stream tables behave exactly as before.
+ *
+ *
+ * {@code execution.runtime-mode} can only be set when a table environment is instantiated, so
+ * each mode gets its own environment over the same catalog instance.
+ */
+public class LakeStreamRoutingITCase {
+
+ @TempDir java.nio.file.Path tempDir;
+
+ private Catalog catalog;
+
+ @BeforeEach
+ public void beforeEach() {
+ FakeStoreCatalogFactory.reset();
+ }
+
+ @AfterEach
+ public void afterEach() throws Exception {
+ if (catalog != null) {
+ catalog.close();
+ }
+ }
+
+ // ------------------------------------------------------------------------
+ // helpers
+ // ------------------------------------------------------------------------
+
+ private void createCatalog(boolean withStore) {
+ Options conf = new Options();
+ conf.setString(CatalogOptions.WAREHOUSE.key(), tempDir.toString());
+ if (withStore) {
+ conf.setString(
+ CatalogOptions.LAKE_STREAM_STORE.key(), FakeStoreCatalogFactory.IDENTIFIER);
+ conf.setString("fakestore.bootstrap.servers", "localhost:9123");
+ }
+ catalog =
+ new FlinkCatalogFactory()
+ .createCatalog(
+ new FactoryUtil.DefaultCatalogContext(
+ "paimon",
+ conf.toMap(),
+ new Configuration(),
+ getClass().getClassLoader()));
+ }
+
+ private StreamTableEnvironment newTableEnv(RuntimeExecutionMode mode) {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment(1);
+ env.setRuntimeMode(mode);
+ StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
+ tEnv.registerCatalog("paimon", catalog);
+ tEnv.useCatalog("paimon");
+ return tEnv;
+ }
+
+ private StreamTableEnvironment newBatchEnvWithLakeStreamTable(String name) {
+ createCatalog(true);
+ StreamTableEnvironment tEnv = newTableEnv(RuntimeExecutionMode.BATCH);
+ tEnv.executeSql(
+ String.format(
+ "CREATE TABLE %s (id INT) WITH ("
+ + "'bucket' = '-1', 'lakestream.enabled' = 'true')",
+ name));
+ return tEnv;
+ }
+
+ private static List queryIds(StreamTableEnvironment tEnv, String sql)
+ throws Exception {
+ List result = new ArrayList<>();
+ try (CloseableIterator it = tEnv.executeSql(sql).collect()) {
+ while (it.hasNext()) {
+ result.add((Integer) it.next().getField(0));
+ }
+ }
+ return result;
+ }
+
+ // ------------------------------------------------------------------------
+ // tests
+ // ------------------------------------------------------------------------
+
+ @Test
+ public void testRegularWriteRoutesToStoreAndOriginalNameReadsPaimonOnly() throws Exception {
+ StreamTableEnvironment batch = newBatchEnvWithLakeStreamTable("t1");
+ StreamTableEnvironment streaming = newTableEnv(RuntimeExecutionMode.STREAMING);
+
+ streaming.executeSql("INSERT INTO t1 SELECT * FROM (VALUES (1), (2)) AS t(id)").await();
+
+ // the regular write went to the fake store, through the delegated sink context
+ assertThat(FakeStoreTableFactory.sinkContexts()).hasSize(1);
+ DynamicTableFactory.Context delegated = FakeStoreTableFactory.sinkContexts().get(0);
+ assertThat(delegated.getObjectIdentifier().getObjectName()).isEqualTo("t1");
+ assertThat(FakeStoreData.rows("default.t1")).hasSize(2);
+
+ // reading the ORIGINAL name accesses Paimon only, which received nothing
+ assertThat(queryIds(batch, "SELECT id FROM t1")).isEmpty();
+ }
+
+ @Test
+ public void testRealTimeReadRoutesToStoreWithBaseIdentifier() throws Exception {
+ StreamTableEnvironment batch = newBatchEnvWithLakeStreamTable("t1");
+ StreamTableEnvironment streaming = newTableEnv(RuntimeExecutionMode.STREAMING);
+
+ streaming.executeSql("INSERT INTO t1 SELECT * FROM (VALUES (1), (2)) AS t(id)").await();
+
+ // the '$rt' read is served by the store's union read
+ assertThat(queryIds(batch, "SELECT id FROM t1$rt")).containsExactlyInAnyOrder(1, 2);
+
+ // the store's source factory saw the BASE identifier, without the '$rt' suffix
+ assertThat(FakeStoreTableFactory.sourceContexts()).isNotEmpty();
+ for (DynamicTableFactory.Context ctx : FakeStoreTableFactory.sourceContexts()) {
+ assertThat(ctx.getObjectIdentifier().getObjectName()).isEqualTo("t1");
+ assertThat(ctx.getObjectIdentifier().getDatabaseName()).isEqualTo("default");
+ }
+ }
+
+ @Test
+ public void testInsertOverwriteIsRejected() throws Exception {
+ StreamTableEnvironment batch = newBatchEnvWithLakeStreamTable("t1");
+
+ // a write to a lake-stream table is served by the store, and the store's sink cannot
+ // execute an overwrite, so the planner rejects the statement rather than let it run
+ // against the wrong side
+ assertThatThrownBy(
+ () ->
+ batch.executeSql(
+ "INSERT OVERWRITE t1 SELECT * FROM (VALUES (5)) AS t(id)"))
+ .hasStackTraceContaining("INSERT OVERWRITE");
+ assertThat(FakeStoreData.rows("default.t1")).isEmpty();
+ }
+
+ @Test
+ public void testOriginalNameReadSucceedsWithoutStoreConfigured() throws Exception {
+ createCatalog(false);
+ StreamTableEnvironment batch = newTableEnv(RuntimeExecutionMode.BATCH);
+ batch.executeSql("CREATE TABLE t3 (id INT) WITH ('bucket' = '-1')");
+
+ batch.executeSql("INSERT INTO t3 SELECT * FROM (VALUES (7)) AS t(id)").await();
+
+ assertThat(queryIds(batch, "SELECT id FROM t3")).containsExactly(7);
+ assertThat(FakeStoreCatalogFactory.createdCatalogs()).isEmpty();
+ assertThat(FakeStoreTableFactory.sinkContexts()).isEmpty();
+ }
+
+ @Test
+ public void testRealTimeNameIsReadOnly() {
+ StreamTableEnvironment streaming = newBatchEnvWithLakeStreamTable("t1");
+
+ // the '$rt' name is read-only; a write has to name the table itself
+ assertThatThrownBy(() -> streaming.executeSql("INSERT INTO `t1$rt` SELECT 1"))
+ .hasStackTraceContaining("read-only view")
+ .hasStackTraceContaining("paimon.default.t1");
+ }
+
+ @Test
+ public void testQueryHintCannotMoveAWriteOffTheStore() throws Exception {
+ StreamTableEnvironment batch = newBatchEnvWithLakeStreamTable("t4");
+ StreamTableEnvironment streaming = newTableEnv(RuntimeExecutionMode.STREAMING);
+
+ streaming
+ .executeSql(
+ "INSERT INTO t4 /*+ OPTIONS('lakestream.enabled' = 'false') */ "
+ + "SELECT * FROM (VALUES (1)) AS t(id)")
+ .await();
+
+ // the persisted option decides the route; a per-statement hint must not move a table's
+ // writes away from the store that holds its recent data
+ assertThat(FakeStoreData.rows("default.t4")).hasSize(1);
+ assertThat(queryIds(batch, "SELECT id FROM t4")).isEmpty();
+ }
+
+ @Test
+ public void testQueryHintCannotMakeAPlainTableWriteToTheStore() throws Exception {
+ createCatalog(true);
+ StreamTableEnvironment batch = newTableEnv(RuntimeExecutionMode.BATCH);
+ batch.executeSql("CREATE TABLE t5 (id INT) WITH ('bucket' = '-1')");
+ StreamTableEnvironment streaming = newTableEnv(RuntimeExecutionMode.STREAMING);
+
+ streaming
+ .executeSql(
+ "INSERT INTO t5 /*+ OPTIONS('lakestream.enabled' = 'true') */ "
+ + "SELECT * FROM (VALUES (1)) AS t(id)")
+ .await();
+
+ assertThat(FakeStoreData.rows("default.t5")).isEmpty();
+ assertThat(queryIds(batch, "SELECT id FROM t5")).containsExactly(1);
+ }
+
+ @Test
+ public void testNonLakeStreamTableBehavesAsBefore() throws Exception {
+ createCatalog(true);
+ StreamTableEnvironment batch = newTableEnv(RuntimeExecutionMode.BATCH);
+ batch.executeSql("CREATE TABLE t2 (id INT) WITH ('bucket' = '-1')");
+
+ batch.executeSql("INSERT INTO t2 SELECT * FROM (VALUES (3)) AS t(id)").await();
+
+ // the write went to Paimon, not to the store
+ assertThat(queryIds(batch, "SELECT id FROM t2")).containsExactly(3);
+ assertThat(FakeStoreData.rows("default.t2")).isEmpty();
+ assertThat(FakeStoreTableFactory.sinkContexts()).isEmpty();
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lakestream/LakeStreamTableFactoryTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lakestream/LakeStreamTableFactoryTest.java
new file mode 100644
index 000000000000..747a9e2b3fce
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lakestream/LakeStreamTableFactoryTest.java
@@ -0,0 +1,629 @@
+/*
+ * 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.flink.lakestream;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.flink.FlinkCatalog;
+import org.apache.paimon.flink.FlinkCatalogFactory;
+import org.apache.paimon.flink.FlinkCatalogOptions;
+import org.apache.paimon.flink.FlinkTableFactory;
+import org.apache.paimon.flink.sink.FlinkTableSink;
+import org.apache.paimon.flink.source.DataTableSource;
+import org.apache.paimon.options.CatalogOptions;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.table.FormatTable;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogDatabaseImpl;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.ResolvedCatalogTable;
+import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.catalog.UniqueConstraint;
+import org.apache.flink.table.catalog.WatermarkSpec;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.sink.SinkV2Provider;
+import org.apache.flink.table.connector.sink.abilities.SupportsDeletePushDown;
+import org.apache.flink.table.connector.sink.abilities.SupportsPartitioning;
+import org.apache.flink.table.connector.sink.abilities.SupportsRowLevelDelete;
+import org.apache.flink.table.connector.sink.abilities.SupportsRowLevelUpdate;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.expressions.ResolvedExpression;
+import org.apache.flink.table.factories.DynamicTableFactory;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests for the lake-stream routing in {@link org.apache.paimon.flink.AbstractFlinkTableFactory}
+ * (PIP-39 source creation L252-281 and sink creation L283-350), exercised through {@link
+ * FlinkTableFactory} created by {@link FlinkCatalog#getFactory()} with the fake store.
+ */
+public class LakeStreamTableFactoryTest {
+
+ private static final ObjectPath T1 = new ObjectPath("db1", "t1");
+ private static final ObjectPath T1_RT = new ObjectPath("db1", "t1$rt");
+ private static final ObjectPath T2 = new ObjectPath("db1", "t2");
+
+ @TempDir java.nio.file.Path tempDir;
+
+ @BeforeEach
+ public void beforeEach() {
+ FakeStoreCatalogFactory.reset();
+ }
+
+ // ------------------------------------------------------------------------
+ // helpers
+ // ------------------------------------------------------------------------
+
+ private Options conf(String factoryMode) {
+ Options conf = new Options();
+ conf.setString(CatalogOptions.WAREHOUSE.key(), tempDir.toString());
+ conf.setString(CatalogOptions.LAKE_STREAM_STORE.key(), FakeStoreCatalogFactory.IDENTIFIER);
+ conf.setString("fakestore.bootstrap.servers", "localhost:9123");
+ if (factoryMode != null) {
+ conf.setString("fakestore." + FakeStoreCatalog.FACTORY_MODE_OPTION, factoryMode);
+ }
+ return conf;
+ }
+
+ private FlinkCatalog newCatalog(String factoryMode) {
+ return new FlinkCatalogFactory()
+ .createCatalog(
+ new FactoryUtil.DefaultCatalogContext(
+ "test-catalog",
+ conf(factoryMode).toMap(),
+ new Configuration(),
+ getClass().getClassLoader()));
+ }
+
+ /** A catalog over the same warehouse that was created without {@code lakestream.store}. */
+ private FlinkCatalog newCatalogWithoutStore() {
+ Options conf = new Options();
+ conf.setString(CatalogOptions.WAREHOUSE.key(), tempDir.toString());
+ return new FlinkCatalogFactory()
+ .createCatalog(
+ new FactoryUtil.DefaultCatalogContext(
+ "test-catalog",
+ conf.toMap(),
+ new Configuration(),
+ getClass().getClassLoader()));
+ }
+
+ private static ResolvedSchema nullableIdSchema() {
+ return new ResolvedSchema(
+ Collections.singletonList(Column.physical("id", DataTypes.INT())),
+ Collections.emptyList(),
+ null);
+ }
+
+ private static ResolvedSchema idSchema() {
+ return new ResolvedSchema(
+ Collections.singletonList(Column.physical("id", DataTypes.INT().notNull())),
+ Collections.emptyList(),
+ UniqueConstraint.primaryKey("PK_id", Collections.singletonList("id")));
+ }
+
+ private void createTable(
+ org.apache.flink.table.catalog.Catalog catalog,
+ ObjectPath path,
+ boolean lakeStreamEnabled)
+ throws Exception {
+ catalog.createDatabase(
+ path.getDatabaseName(), new CatalogDatabaseImpl(new HashMap<>(), null), true);
+ Map options = new HashMap<>();
+ options.put(CoreOptions.BUCKET.key(), "1");
+ if (lakeStreamEnabled) {
+ options.put(CoreOptions.LAKE_STREAM_ENABLED.key(), "true");
+ }
+ ResolvedSchema resolvedSchema = idSchema();
+ CatalogTable table =
+ new ResolvedCatalogTable(
+ CatalogTable.newBuilder()
+ .schema(
+ Schema.newBuilder()
+ .fromResolvedSchema(resolvedSchema)
+ .build())
+ .partitionKeys(Collections.emptyList())
+ .options(options)
+ .build(),
+ resolvedSchema);
+ catalog.createTable(path, table, false);
+ }
+
+ private static ResolvedCatalogTable resolved(CatalogBaseTable table, ResolvedSchema schema) {
+ return new ResolvedCatalogTable((CatalogTable) table, schema);
+ }
+
+ /** A {@code null} table is for a decision that is reached before the table is asked for. */
+ private DynamicTableFactory.Context context(
+ ObjectIdentifier identifier, ResolvedCatalogTable table) {
+ return context(identifier, table, new Configuration());
+ }
+
+ private DynamicTableFactory.Context context(
+ ObjectIdentifier identifier, ResolvedCatalogTable table, ReadableConfig configuration) {
+ return new DynamicTableFactory.Context() {
+ @Override
+ public ObjectIdentifier getObjectIdentifier() {
+ return identifier;
+ }
+
+ @Override
+ public ResolvedCatalogTable getCatalogTable() {
+ if (table == null) {
+ throw new UnsupportedOperationException();
+ }
+ return table;
+ }
+
+ @Override
+ public boolean isTemporary() {
+ return false;
+ }
+
+ @Override
+ public ReadableConfig getConfiguration() {
+ return configuration;
+ }
+
+ @Override
+ public ClassLoader getClassLoader() {
+ return getClass().getClassLoader();
+ }
+ };
+ }
+
+ private static ObjectIdentifier id(String objectName) {
+ return ObjectIdentifier.of("test-catalog", "db1", objectName);
+ }
+
+ private static DynamicTableSink.Context sinkContext(boolean bounded) {
+ return new DynamicTableSink.Context() {
+ @Override
+ public boolean isBounded() {
+ return bounded;
+ }
+
+ @Override
+ public TypeInformation createTypeInformation(DataType dataType) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public TypeInformation createTypeInformation(LogicalType logicalType) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public DynamicTableSink.DataStructureConverter createDataStructureConverter(
+ DataType dataType) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Optional getTargetColumns() {
+ return Optional.empty();
+ }
+ };
+ }
+
+ // ------------------------------------------------------------------------
+ // source routing (PIP-39 L252-281)
+ // ------------------------------------------------------------------------
+
+ @Test
+ public void testRealTimeReadRoutesToStoreSourceFactory() throws Exception {
+ FlinkCatalog catalog = newCatalog(null);
+ createTable(catalog, T1, true);
+ // the planner resolves 't1$rt' through the wrapper, which returns the store table
+ ResolvedCatalogTable storeTable = resolved(catalog.getTable(T1_RT), nullableIdSchema());
+ FlinkTableFactory factory = (FlinkTableFactory) catalog.getFactory().get();
+
+ DynamicTableSource source =
+ factory.createDynamicTableSource(context(id("t1$rt"), storeTable));
+
+ assertThat(source).isInstanceOf(FakeStoreSource.class);
+ assertThat(FakeStoreTableFactory.sourceContexts()).hasSize(1);
+ DynamicTableFactory.Context delegated = FakeStoreTableFactory.sourceContexts().get(0);
+ // delegated with the BASE identifier, without the '$rt' suffix
+ assertThat(delegated.getObjectIdentifier().getObjectName()).isEqualTo("t1");
+ assertThat(delegated.getObjectIdentifier().getDatabaseName()).isEqualTo("db1");
+ // the store table's options
+ assertThat(delegated.getCatalogTable().getOptions())
+ .containsEntry("connector", "fakestore");
+ // the store connection options after prefix removal
+ assertThat(delegated.getConfiguration().get(FakeStoreCatalogFactory.BOOTSTRAP_SERVERS))
+ .isEqualTo("localhost:9123");
+ catalog.close();
+ }
+
+ @Test
+ public void testQueryHintsReachTheStoreSourceFactory() throws Exception {
+ FlinkCatalog catalog = newCatalog(null);
+ createTable(catalog, T1, true);
+ // the planner applies '/*+ OPTIONS(...) */' on top of the table the catalog resolved, and a
+ // real store needs those hints to reach its source factory, e.g. to bound a log scan
+ CatalogBaseTable storeTable = catalog.getTable(T1_RT);
+ Map hintedOptions = new HashMap<>(storeTable.getOptions());
+ hintedOptions.put("scan.startup.mode", "earliest");
+ ResolvedCatalogTable hinted =
+ resolved(((CatalogTable) storeTable).copy(hintedOptions), nullableIdSchema());
+ FlinkTableFactory factory = (FlinkTableFactory) catalog.getFactory().get();
+
+ factory.createDynamicTableSource(context(id("t1$rt"), hinted));
+
+ DynamicTableFactory.Context delegated = FakeStoreTableFactory.sourceContexts().get(0);
+ assertThat(delegated.getCatalogTable().getOptions())
+ .containsEntry("connector", "fakestore")
+ .containsEntry("scan.startup.mode", "earliest");
+ catalog.close();
+ }
+
+ @Test
+ public void testOriginalNameReadCreatesPaimonSource() throws Exception {
+ FlinkCatalog catalog = newCatalog(null);
+ createTable(catalog, T1, true);
+ ResolvedCatalogTable paimonTable = resolved(catalog.getTable(T1), idSchema());
+ FlinkTableFactory factory = (FlinkTableFactory) catalog.getFactory().get();
+
+ DynamicTableSource source =
+ factory.createDynamicTableSource(context(id("t1"), paimonTable));
+
+ assertThat(source).isInstanceOf(DataTableSource.class);
+ assertThat(FakeStoreTableFactory.sourceContexts()).isEmpty();
+ catalog.close();
+ }
+
+ @Test
+ public void testRealTimeReadOnNonLakeStreamTableIsNotRouted() throws Exception {
+ FlinkCatalog catalog = newCatalog(null);
+ createTable(catalog, T2, false);
+ ResolvedCatalogTable paimonTable = resolved(catalog.getTable(T2), idSchema());
+ FlinkTableFactory factory = (FlinkTableFactory) catalog.getFactory().get();
+
+ DynamicTableSource source =
+ factory.createDynamicTableSource(context(id("t2$rt"), paimonTable));
+
+ // the wrapper already rejects 't2$rt' resolution; at factory level the gate must not route
+ // a '$rt' identifier whose base table has lakestream.enabled = false
+ assertThat(source).isNotInstanceOf(FakeStoreSource.class);
+ assertThat(FakeStoreTableFactory.sourceContexts()).isEmpty();
+ catalog.close();
+ }
+
+ @Test
+ public void testRealTimeReadWithBranchIsNotRouted() throws Exception {
+ FlinkCatalog catalog = newCatalog(null);
+ createTable(catalog, T1, true);
+ ResolvedCatalogTable paimonTable = resolved(catalog.getTable(T1), idSchema());
+ FlinkTableFactory factory = (FlinkTableFactory) catalog.getFactory().get();
+
+ DynamicTableSource source =
+ factory.createDynamicTableSource(context(id("t1$branch_b1$rt"), paimonTable));
+
+ // a branch access must never be misrouted to the store ('endsWith($rt)' would misroute it)
+ assertThat(source).isNotInstanceOf(FakeStoreSource.class);
+ assertThat(FakeStoreTableFactory.sourceContexts()).isEmpty();
+ catalog.close();
+ }
+
+ @Test
+ public void testRealTimeReadInConnectorModeIsNotRouted() throws Exception {
+ FlinkCatalog catalog = newCatalog(null);
+ createTable(catalog, T1, true);
+ ResolvedCatalogTable paimonTable = resolved(catalog.getTable(T1), idSchema());
+ // connector mode: the factory is discovered by identifier and gets no catalog at all
+ FlinkTableFactory factory = new FlinkTableFactory();
+
+ DynamicTableSource source =
+ factory.createDynamicTableSource(context(id("t1$rt"), paimonTable));
+
+ assertThat(source).isNotInstanceOf(FakeStoreSource.class);
+ assertThat(FakeStoreTableFactory.sourceContexts()).isEmpty();
+ catalog.close();
+ }
+
+ @Test
+ public void testRealTimeReadFailsWhenStoreExposesNoSourceFactory() throws Exception {
+ FlinkCatalog catalog = newCatalog("sink-only");
+ createTable(catalog, T1, true);
+ ResolvedCatalogTable storeTable = resolved(catalog.getTable(T1_RT), nullableIdSchema());
+ FlinkTableFactory factory = (FlinkTableFactory) catalog.getFactory().get();
+
+ assertThatThrownBy(() -> factory.createDynamicTableSource(context(id("t1$rt"), storeTable)))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("fakestore");
+ catalog.close();
+ }
+
+ // ------------------------------------------------------------------------
+ // sink routing (PIP-39 L283-350)
+ // ------------------------------------------------------------------------
+
+ @Test
+ public void testRegularWriteGetsTheStoreSink() throws Exception {
+ FlinkCatalog catalog = newCatalog(null);
+ createTable(catalog, T1, true);
+ ResolvedCatalogTable paimonTable = resolved(catalog.getTable(T1), idSchema());
+ FlinkTableFactory factory = (FlinkTableFactory) catalog.getFactory().get();
+
+ DynamicTableSink sink = factory.createDynamicTableSink(context(id("t1"), paimonTable));
+
+ // a regular write is executed by the store, so the planner is handed the store's own sink
+ // and discovers its abilities on it directly
+ assertThat(sink).isInstanceOf(FakeStoreSink.class);
+ assertThat(sink).isNotInstanceOf(FlinkTableSink.class);
+
+ // the store sink is built through the delegated context
+ assertThat(FakeStoreTableFactory.sinkContexts()).hasSize(1);
+ DynamicTableFactory.Context delegated = FakeStoreTableFactory.sinkContexts().get(0);
+ // regular writes keep the ORIGINAL identifier (no '$rt' involved)
+ assertThat(delegated.getObjectIdentifier().getObjectName()).isEqualTo("t1");
+ assertThat(delegated.getCatalogTable().getOptions())
+ .containsEntry("connector", "fakestore");
+ assertThat(delegated.getConfiguration().get(FakeStoreCatalogFactory.BOOTSTRAP_SERVERS))
+ .isEqualTo("localhost:9123");
+
+ assertThat(sink.asSummaryString()).isEqualTo("FakeStoreSink");
+ assertThat(sink.getSinkRuntimeProvider(sinkContext(true)))
+ .isInstanceOf(SinkV2Provider.class);
+ catalog.close();
+ }
+
+ @Test
+ public void testRowLevelOperationsAreForwardedToTheStore() throws Exception {
+ FlinkCatalog catalog = newCatalog(null);
+ createTable(catalog, T1, true);
+ ResolvedCatalogTable paimonTable = resolved(catalog.getTable(T1), idSchema());
+ FlinkTableFactory factory = (FlinkTableFactory) catalog.getFactory().get();
+ FakeStoreSink.resetAbilityCalls();
+
+ DynamicTableSink sink = factory.createDynamicTableSink(context(id("t1"), paimonTable));
+
+ // a DELETE/UPDATE is a regular write, so the store executes it
+ ((SupportsRowLevelDelete) sink).applyRowLevelDelete(null);
+ ((SupportsRowLevelUpdate) sink)
+ .applyRowLevelUpdate(
+ Collections.singletonList(Column.physical("id", DataTypes.INT())), null);
+ ((SupportsDeletePushDown) sink)
+ .applyDeleteFilters(Collections.emptyList());
+ ((SupportsDeletePushDown) sink).executeDeletion();
+
+ assertThat(FakeStoreSink.abilityCalls())
+ .contains(
+ "applyRowLevelDelete",
+ "applyRowLevelUpdate(1)",
+ "applyDeleteFilters(0)",
+ "executeDeletion");
+ catalog.close();
+ }
+
+ @Test
+ public void testStaticPartitionReachesTheStoreSink() throws Exception {
+ FlinkCatalog catalog = newCatalog(null);
+ createTable(catalog, T1, true);
+ ResolvedCatalogTable paimonTable = resolved(catalog.getTable(T1), idSchema());
+ FlinkTableFactory factory = (FlinkTableFactory) catalog.getFactory().get();
+ FakeStoreSink.resetAbilityCalls();
+
+ DynamicTableSink sink = factory.createDynamicTableSink(context(id("t1"), paimonTable));
+
+ Map partition = new HashMap<>();
+ partition.put("dt", "20261019");
+
+ // a regular write is executed by the store, so the partition spec is applied to its sink
+ ((SupportsPartitioning) sink).applyStaticPartition(partition);
+ assertThat(FakeStoreSink.abilityCalls()).contains("applyStaticPartition({dt=20261019})");
+ catalog.close();
+ }
+
+ @Test
+ public void testFormatTableCannotBeALakeStreamTable() throws Exception {
+ // a format table has no Paimon history for the store to tier into, so it is rejected on
+ // both paths rather than left half routed
+ FormatTable formatTable = mock(FormatTable.class);
+ when(formatTable.options())
+ .thenReturn(
+ Collections.singletonMap(CoreOptions.LAKE_STREAM_ENABLED.key(), "true"));
+ Catalog paimonCatalog = mock(Catalog.class);
+ when(paimonCatalog.getTable(any(Identifier.class))).thenReturn(formatTable);
+
+ Options options = new Options();
+ options.set(FlinkCatalogOptions.DISABLE_CREATE_TABLE_IN_DEFAULT_DB, true);
+ FlinkCatalog catalog = new FlinkCatalog(paimonCatalog, "test-catalog", "default", options);
+ LakeStreamTableFactory lakeStream = new LakeStreamTableFactory(catalog);
+
+ assertThatThrownBy(() -> lakeStream.createDynamicTableSink(context(id("t1"), null)))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("format tables")
+ .hasMessageContaining(CoreOptions.LAKE_STREAM_ENABLED.key());
+ assertThatThrownBy(() -> lakeStream.createDynamicTableSource(context(id("t1$rt"), null)))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("format tables");
+ catalog.close();
+ }
+
+ @Test
+ public void testSinkNotRoutedWhenLakeStreamDisabled() throws Exception {
+ FlinkCatalog catalog = newCatalog(null);
+ createTable(catalog, T2, false);
+ ResolvedCatalogTable paimonTable = resolved(catalog.getTable(T2), idSchema());
+ FlinkTableFactory factory = (FlinkTableFactory) catalog.getFactory().get();
+
+ DynamicTableSink sink = factory.createDynamicTableSink(context(id("t2"), paimonTable));
+
+ assertThat(sink).isInstanceOf(FlinkTableSink.class);
+ assertThat(FakeStoreTableFactory.sinkContexts()).isEmpty();
+ catalog.close();
+ }
+
+ @Test
+ public void testJobLevelOptionCannotMakeTheGuardBlameAMissingStore() throws Exception {
+ FlinkCatalog catalog = newCatalog(null);
+ createTable(catalog, T2, false);
+ ResolvedCatalogTable paimonTable = resolved(catalog.getTable(T2), idSchema());
+ FlinkTableFactory factory = (FlinkTableFactory) catalog.getFactory().get();
+
+ // a job-level dynamic option reaches the built table but not the persisted one, so the
+ // write stays on Paimon. The missing-store guard must stay quiet: this catalog has a store,
+ // and blaming its absence would be a false diagnosis.
+ Configuration configuration = new Configuration();
+ configuration.setString(
+ "paimon.test-catalog.db1.t2." + CoreOptions.LAKE_STREAM_ENABLED.key(), "true");
+
+ DynamicTableSink sink =
+ factory.createDynamicTableSink(context(id("t2"), paimonTable, configuration));
+
+ assertThat(sink).isInstanceOf(FlinkTableSink.class);
+ assertThat(FakeStoreTableFactory.sinkContexts()).isEmpty();
+ catalog.close();
+ }
+
+ @Test
+ public void testConnectorModeCannotBypassTheStore() throws Exception {
+ FlinkCatalog catalog = newCatalog(null);
+ createTable(catalog, T1, true);
+ ResolvedCatalogTable paimonTable = resolved(catalog.getTable(T1), idSchema());
+ // connector mode: the factory is discovered by identifier and gets no catalog at all, so it
+ // has no store to route to
+ FlinkTableFactory factory = new FlinkTableFactory();
+
+ assertThatThrownBy(() -> factory.createDynamicTableSink(context(id("t1"), paimonTable)))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining(CatalogOptions.LAKE_STREAM_STORE.key())
+ .hasMessageContaining(CoreOptions.LAKE_STREAM_ENABLED.key());
+ assertThat(FakeStoreTableFactory.sinkContexts()).isEmpty();
+ catalog.close();
+ }
+
+ @Test
+ public void testWriteToLakeStreamTableWithoutStoreConfiguredIsRejected() throws Exception {
+ FlinkCatalog withStore = newCatalog(null);
+ createTable(withStore, T1, true);
+ withStore.close();
+
+ // the same warehouse reached through a catalog that was created without 'lakestream.store'
+ FlinkCatalog withoutStore = newCatalogWithoutStore();
+ ResolvedCatalogTable paimonTable = resolved(withoutStore.getTable(T1), idSchema());
+ FlinkTableFactory factory = (FlinkTableFactory) withoutStore.getFactory().get();
+
+ // writing to Paimon here would leave the store serving stale-plus-diverged data for the
+ // same keys, so the missing catalog option is reported instead
+ assertThatThrownBy(() -> factory.createDynamicTableSink(context(id("t1"), paimonTable)))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining(CatalogOptions.LAKE_STREAM_STORE.key())
+ .hasMessageContaining("test-catalog.db1.t1");
+ assertThat(FakeStoreTableFactory.sinkContexts()).isEmpty();
+ withoutStore.close();
+ }
+
+ @Test
+ public void testWriteToPlainTableWithoutStoreConfiguredIsAllowed() throws Exception {
+ FlinkCatalog withoutStore = newCatalogWithoutStore();
+ createTable(withoutStore, T2, false);
+ ResolvedCatalogTable paimonTable = resolved(withoutStore.getTable(T2), idSchema());
+ FlinkTableFactory factory = (FlinkTableFactory) withoutStore.getFactory().get();
+
+ DynamicTableSink sink = factory.createDynamicTableSink(context(id("t2"), paimonTable));
+
+ assertThat(sink).isInstanceOf(FlinkTableSink.class);
+ assertThat(FakeStoreTableFactory.sinkContexts()).isEmpty();
+ withoutStore.close();
+ }
+
+ @Test
+ public void testSinkFailsWhenStoreExposesNoSinkFactory() throws Exception {
+ FlinkCatalog catalog = newCatalog("source-only");
+ createTable(catalog, T1, true);
+ ResolvedCatalogTable paimonTable = resolved(catalog.getTable(T1), idSchema());
+ FlinkTableFactory factory = (FlinkTableFactory) catalog.getFactory().get();
+
+ assertThatThrownBy(() -> factory.createDynamicTableSink(context(id("t1"), paimonTable)))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("fakestore");
+ catalog.close();
+ }
+
+ @Test
+ public void testSinkFailsClearlyWhenTheStoreSideIsMissing() throws Exception {
+ FlinkCatalog catalog = newCatalog(null);
+ createTable(catalog, T1, true);
+ FakeStoreCatalog.markTableMissing(T1);
+ FlinkTableFactory factory = (FlinkTableFactory) catalog.getFactory().get();
+ ResolvedCatalogTable paimonTable = resolved(catalog.getTable(T1), idSchema());
+
+ assertThatThrownBy(() -> factory.createDynamicTableSink(context(id("t1"), paimonTable)))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("does not exist in the Lakestream Store")
+ .hasMessageContaining(FakeStoreCatalogFactory.IDENTIFIER);
+ catalog.close();
+ }
+
+ @Test
+ public void testWriteThroughTheRealTimeNameIsRejected() throws Exception {
+ FlinkCatalog catalog = newCatalog(null);
+ createTable(catalog, T1, true);
+ FlinkTableFactory factory = (FlinkTableFactory) catalog.getFactory().get();
+ ResolvedCatalogTable storeTable = resolved(catalog.getTable(T1_RT), idSchema());
+
+ assertThatThrownBy(() -> factory.createDynamicTableSink(context(id("t1$rt"), storeTable)))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("read-only view")
+ .hasMessageContaining("test-catalog.db1.t1");
+ catalog.close();
+ }
+
+ @Test
+ public void testSinkFailsWhenStoreExposesNoFactoryAtAll() throws Exception {
+ FlinkCatalog catalog = newCatalog("none");
+ createTable(catalog, T1, true);
+ ResolvedCatalogTable paimonTable = resolved(catalog.getTable(T1), idSchema());
+ FlinkTableFactory factory = (FlinkTableFactory) catalog.getFactory().get();
+
+ assertThatThrownBy(() -> factory.createDynamicTableSink(context(id("t1"), paimonTable)))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("fakestore");
+ catalog.close();
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/test/resources/META-INF/services/org.apache.flink.table.factories.Factory b/paimon-flink/paimon-flink-common/src/test/resources/META-INF/services/org.apache.flink.table.factories.Factory
new file mode 100644
index 000000000000..9d0b5be8c78d
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/test/resources/META-INF/services/org.apache.flink.table.factories.Factory
@@ -0,0 +1,16 @@
+# 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.
+
+org.apache.paimon.flink.lakestream.FakeStoreCatalogFactory
diff --git a/pom.xml b/pom.xml
index d6255a559728..12881012f962 100644
--- a/pom.xml
+++ b/pom.xml
@@ -128,6 +128,9 @@ under the License.
3.4.0-1.20
1.20.0
8.0.27
+ 1.0.0
+ 5.4.0
+ 3.7.1
paimon-spark3-common_${scala.binary.version}