yunfengzhou-hub commented on code in PR #10833: URL: https://github.com/apache/paimon/pull/10833#discussion_r4409819663
########## 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: Fixed by refusing the statement rather than routing the scan. LakeStreamDynamicTableSink wraps the store's sink and throws on applyRowLevelUpdate and applyRowLevelDelete: the scan resolves to Paimon snapshots while the modification goes to the store, so it could miss rows that have not been tiered or compute from a stale value. SupportsOverwrite and SupportsDeletePushDown are withheld as well — a DELETE then falls back to the refused row-level path instead of being pushed into the store — while SupportsPartitioning is still forwarded, and LakeStreamSinkAbilityDriftTest keeps the withheld set in step with the Paimon and Fluss sinks. Which name should expose a view that is both readable and writable is a PIP-39 question, so that is left to a follow-up. -- 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]
