Copilot commented on code in PR #10833: URL: https://github.com/apache/paimon/pull/10833#discussion_r4396105640
########## paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lakestream/LakeStreamTableFactory.java: ########## @@ -0,0 +1,304 @@ +/* + * 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.AbstractFlinkTableFactory; +import org.apache.paimon.flink.FlinkCatalog; +import org.apache.paimon.flink.FlinkCatalogFactory; +import org.apache.paimon.options.CatalogOptions; +import org.apache.paimon.options.Options; +import org.apache.paimon.table.FormatTable; +import org.apache.paimon.table.Table; + +import org.apache.flink.configuration.ConfigOption; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.ReadableConfig; +import org.apache.flink.table.api.ValidationException; +import org.apache.flink.table.catalog.CatalogBaseTable; +import org.apache.flink.table.catalog.CatalogTable; +import org.apache.flink.table.catalog.ObjectIdentifier; +import org.apache.flink.table.catalog.ResolvedCatalogTable; +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.factories.Factory; + +import javax.annotation.Nullable; + +import java.util.Collections; +import java.util.Map; +import java.util.Optional; +import java.util.Set; + +/** + * Builds the source and the sink of a lake-stream table by delegating to the Lakestream Store's own + * table factory. Not registered with Flink's factory SPI: {@link #factoryIdentifier} exists because + * the interface asks for it, and nothing discovers this object by it. + */ +public class LakeStreamTableFactory implements DynamicTableSourceFactory, DynamicTableSinkFactory { + + private final FlinkCatalog catalog; + + public LakeStreamTableFactory(FlinkCatalog catalog) { + this.catalog = catalog; + } + + @Override + public String factoryIdentifier() { + return FlinkCatalogFactory.IDENTIFIER; + } + + @Override + public Set<ConfigOption<?>> requiredOptions() { + return Collections.emptySet(); + } + + @Override + public Set<ConfigOption<?>> optionalOptions() { + return Collections.emptySet(); + } + + /** Whether the read is a {@code $rt} read of a lake-stream table. */ + public static boolean isLakeStreamRead( + Catalog paimonCatalog, DynamicTableFactory.Context context) { + return realTimeReadBaseTable(paimonCatalog, context) != null; + } + + /** Whether the write is to a lake-stream table, or through the read-only {@code $rt} name. */ + public static boolean isLakeStreamWrite( + Catalog paimonCatalog, DynamicTableFactory.Context context) { + Identifier identifier = identifierOf(context); + return LakeStreamUtils.isRealTimeTable(identifier) + || persistedLakeStreamTable(paimonCatalog, identifier) != null; + } + + @Override + public DynamicTableSource createDynamicTableSource(DynamicTableFactory.Context context) { + Table baseTable = realTimeReadBaseTable(catalog.catalog(), context); + ObjectIdentifier baseIdentifier = + LakeStreamUtils.stripRealTimeSuffix(context.getObjectIdentifier()); + checkNotFormatTable(baseTable, baseIdentifier); + org.apache.flink.table.catalog.Catalog storeCatalog = + catalog.getOrCreateLakeStreamStoreCatalog(); + // the catalog already resolved this read against the store, so the planner's table is the + // store's table with any query hints applied on top of it + CatalogBaseTable storeTable = context.getCatalogTable(); + DynamicTableSourceFactory sourceFactory = + storeFactory( + storeCatalog, + DynamicTableSourceFactory.class, + "create a source for '$rt'-suffixed reads"); + return sourceFactory.createDynamicTableSource( + delegatedContext(context, baseIdentifier, storeTable)); + } + + @Override + public DynamicTableSink createDynamicTableSink(DynamicTableFactory.Context context) { + checkNotRealTimeWrite(context); + Table table = persistedLakeStreamTable(catalog.catalog(), identifierOf(context)); + checkNotFormatTable(table, context.getObjectIdentifier()); + ObjectIdentifier identifier = context.getObjectIdentifier(); + org.apache.flink.table.catalog.Catalog storeCatalog = + catalog.getOrCreateLakeStreamStoreCatalog(); + // the store table is fetched from the store catalog because the store's sink factory may + // require provider-specific table options (PIP-39 L289) + CatalogBaseTable storeTable = + catalog.requireLakeStreamStoreTable(identifier.toObjectPath()); + DynamicTableSinkFactory sinkFactory = + storeFactory( + storeCatalog, DynamicTableSinkFactory.class, "create a sink for writes"); + return sinkFactory.createDynamicTableSink( + delegatedContext(context, identifier, storeTable)); Review Comment: Returning the store sink exposes its row-level UPDATE/DELETE abilities, but scan-based modifications still read the original table name through the Paimon source. When the store supports these abilities, modifications can miss untiered rows or compute updates from stale snapshot values. Route modification scans through the store's current view, or reject scan-based UPDATE/DELETE until that is supported, and add an end-to-end SQL test with an untiered row. ########## paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/AbstractFlinkTableFactory.java: ########## @@ -81,12 +82,74 @@ public abstract class AbstractFlinkTableFactory @Nullable private final FlinkCatalog flinkCatalog; + /** Builds the source and sink of a lake-stream table; {@code null} when no store is named. */ + @Nullable private final LakeStreamTableFactory lakeStreamTableFactory; + public AbstractFlinkTableFactory(@Nullable FlinkCatalog flinkCatalog) { + this(flinkCatalog, null); + } + + public AbstractFlinkTableFactory( + @Nullable FlinkCatalog flinkCatalog, + @Nullable LakeStreamTableFactory lakeStreamTableFactory) { this.flinkCatalog = flinkCatalog; + this.lakeStreamTableFactory = lakeStreamTableFactory; } @Override public DynamicTableSource createDynamicTableSource(Context context) { + if (isLakeStreamRead(context)) { + return createLakeStreamSource(context); + } + return createPaimonTableSource(context); + } + + @Override + public DynamicTableSink createDynamicTableSink(Context context) { + if (isLakeStreamWrite(context)) { + return createLakeStreamSink(context); + } + return createPaimonTableSink(context, buildPaimonTable(context)); Review Comment: A connector mapping declared with only `connector='paimon'` and `path=...` bypasses the missing-store guard when the persisted table has `lakestream.enabled=true`, so its writes go directly to Paimon and can diverge from the store. A `lakestream.enabled=false` hint can bypass the guard too. Check the target's persisted schema flag independently of DDL and hint overrides before creating the connector-mode sink, and cover a path-only mapping in a regression test. ########## paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalog.java: ########## @@ -973,10 +1002,153 @@ public final void open() throws CatalogException {} @Override public final void close() throws CatalogException { + Throwable failure = null; try { catalog.close(); } catch (Exception e) { - throw new CatalogException("Failed to close catalog " + catalog.toString(), e); + failure = new CatalogException("Failed to close catalog " + catalog.toString(), e); + } + try { + closeLakeStreamStoreCatalog(); + } catch (Throwable t) { + failure = ExceptionUtils.firstOrSuppressed(t, failure); + } + if (failure != null) { + ExceptionUtils.rethrow(failure); + } + } + + // ------------------------------------------------------------------------ + // Lake-stream mode (PIP-39) + // ------------------------------------------------------------------------ + + /** Whether this catalog names a Lakestream Store. */ + public boolean lakeStreamStoreConfigured() { + return lakeStreamStore() != null; + } + + /** The configured store identifier, or {@code <unset>} when there is none. */ + public String lakeStreamStoreIdentifier() { + String store = lakeStreamStore(); + return store == null ? "<unset>" : store; + } + + /** The trimmed {@code lakestream.store} value, or {@code null} when it is absent or blank. */ + @Nullable + private String lakeStreamStore() { + String store = options.get(CatalogOptions.LAKE_STREAM_STORE); + return StringUtils.isNullOrWhitespaceOnly(store) ? null : store.trim(); + } + + /** + * Lazily discovers, creates, opens and caches the Lakestream Store catalog. Lazy so that an + * unreachable or absent store does not stop this catalog from resolving table metadata. + */ + public synchronized org.apache.flink.table.catalog.Catalog getOrCreateLakeStreamStoreCatalog() { + org.apache.flink.table.catalog.Catalog cached = lakeStreamStoreCatalog; + if (cached != null) { + return cached; + } + + String identifier = lakeStreamStore(); + if (identifier == null) { + throw new CatalogException( + String.format( + "Lake-stream mode requires the catalog option '%s' to name the " + + "Lakestream Store implementation (e.g. '%s' = 'fluss'), but " + + "catalog '%s' does not set it to a store identifier.", + CatalogOptions.LAKE_STREAM_STORE.key(), + CatalogOptions.LAKE_STREAM_STORE.key(), + name)); + } + + CatalogFactory catalogFactory; + try { + catalogFactory = + FactoryUtil.discoverFactory( + Thread.currentThread().getContextClassLoader(), Review Comment: Lazy store discovery uses the calling thread's context class loader instead of the class loader supplied to `FlinkCatalogFactory.createCatalog`. Those loaders need not be the same: a store connector visible only through Flink's supplied user-code/catalog loader will fail discovery here despite being correctly installed. Please retain the supplied class loader in `FlinkCatalog` and use it for both `discoverFactory` and the delegated `DefaultCatalogContext`, rather than depending on whichever thread first accesses the store. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
