yunfengzhou-hub commented on code in PR #10833: URL: https://github.com/apache/paimon/pull/10833#discussion_r4467572525
########## paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lakestream/LakeStreamTableFactory.java: ########## @@ -0,0 +1,336 @@ +/* + * 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.CatalogContext; +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.flink.FlinkFileIOLoader; +import org.apache.paimon.options.CatalogOptions; +import org.apache.paimon.options.Options; +import org.apache.paimon.table.FileStoreTableFactory; +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 an SPI-registered factory: {@link #factoryIdentifier} is implemented only + * because the interface requires 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 asks for the Lakestream Store's view of a table. */ + public static boolean isLakeStreamRead( + @Nullable FlinkCatalog catalog, DynamicTableFactory.Context context) { + if (catalog == null) { + // with no catalog there is no lake-stream table to confirm the name against, so the + // name alone counts; the read is then refused for lack of a store rather than served + // from Paimon + return LakeStreamUtils.isRealTimeTable(context.getObjectIdentifier().toObjectPath()); + } + return realTimeReadBaseTable(catalog.catalog(), context) != null; + } + + /** + * Whether the write is to a lake-stream table, or through the read-only {@code $rt} name. + * Connector options and query hints do not change the answer. + */ + public static boolean isLakeStreamWrite( + @Nullable FlinkCatalog catalog, DynamicTableFactory.Context context) { + Identifier identifier = identifierOf(context); + if (LakeStreamUtils.isRealTimeTable(identifier)) { + return true; + } + return catalog != null + ? persistedLakeStreamTable(catalog.catalog(), identifier) != null + : lakeStreamEnabledAtPath(context); + } + + @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 new LakeStreamDynamicTableSink( + sinkFactory.createDynamicTableSink( + delegatedContext(context, identifier, storeTable)), + identifier); + } + + /** A format table has no Paimon history for the store to tier into, so it cannot be one. */ + private void checkNotFormatTable(@Nullable Table table, ObjectIdentifier identifier) { + if (table instanceof FormatTable) { + throw new ValidationException( + String.format( + "Table '%s' cannot be a lake-stream table: '%s' is only supported on " + + "Paimon tables, not on format tables.", + identifier.asSummaryString(), CoreOptions.LAKE_STREAM_ENABLED.key())); + } + } + + /** The base table of a {@code $rt} read, or {@code null} when the read stays on Paimon. */ + @Nullable + private static Table realTimeReadBaseTable( + Catalog paimonCatalog, DynamicTableFactory.Context context) { + Identifier identifier = identifierOf(context); + if (!LakeStreamUtils.isRealTimeTable(identifier)) { + return null; + } + return persistedLakeStreamTable( + paimonCatalog, + Identifier.create(identifier.getDatabaseName(), identifier.getTableName())); + } + + /** The Paimon table when lake-stream mode is enabled on it, otherwise {@code null}. */ + @Nullable + private static Table persistedLakeStreamTable(Catalog paimonCatalog, Identifier identifier) { + try { + Table table = paimonCatalog.getTable(identifier); + return Options.fromMap(table.options()).get(CoreOptions.LAKE_STREAM_ENABLED) + ? table + : null; + } catch (Catalog.TableNotExistException e) { + return null; + } + } + + /** Whether the schema persisted at the connector's {@code path} has lake-stream mode on. */ + private static boolean lakeStreamEnabledAtPath(DynamicTableFactory.Context context) { + Options options = Options.fromMap(context.getCatalogTable().getOptions()); + String path = options.get(CoreOptions.PATH); + if (path == null) { + return options.get(CoreOptions.LAKE_STREAM_ENABLED); + } + // only the location goes in: the factory merges whatever it is given over the persisted + // schema, so passing the connector options would restate the very flag being read + Options location = new Options(); + location.set(CoreOptions.PATH, path); + location.set(CoreOptions.BRANCH, options.get(CoreOptions.BRANCH)); + Table table = + FileStoreTableFactory.create( + CatalogContext.create(location, new FlinkFileIOLoader())); Review Comment: Fixed. The connector mapping's options are now used to construct the FileIO, while only path and branch are passed when loading the table. Therefore lakestream.enabled still comes exclusively from the persisted schema and cannot be overridden by DDL options or hints. -- 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]
