JingsongLi commented on code in PR #10833: URL: https://github.com/apache/paimon/pull/10833#discussion_r4491291903
########## paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lakestream/LakeStreamUtils.java: ########## @@ -0,0 +1,189 @@ +/* + * 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.Identifier; +import org.apache.paimon.flink.FlinkCatalog; +import org.apache.paimon.flink.LogicalTypeConversion; +import org.apache.paimon.schema.Schema; +import org.apache.paimon.table.Table; +import org.apache.paimon.types.DataField; + +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.ObjectPath; +import org.apache.flink.table.types.AbstractDataType; +import org.apache.flink.table.types.DataType; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; + +/** Utilities for lake-stream mode. */ +public class LakeStreamUtils { + + /** The {@code $rt} suffix, e.g. {@code my_table$rt}. */ + public static final String REAL_TIME_TABLE_SUFFIX = "rt"; + + /** + * Whether the identifier carries the {@code $rt} suffix. Branch access such as {@code + * t1$branch_b1$rt} deliberately does not. + */ + public static boolean isRealTimeTable(Identifier identifier) { + return identifier.getBranchName() == null + && REAL_TIME_TABLE_SUFFIX.equals(identifier.getSystemTableName()); + } + + public static boolean isRealTimeTable(ObjectPath tablePath) { + return isRealTimeTable(FlinkCatalog.toIdentifier(tablePath)); + } + + /** Strips the {@code $rt} suffix, keeping catalog and database. */ + public static ObjectIdentifier stripRealTimeSuffix(ObjectIdentifier identifier) { + Identifier parsed = + new Identifier(identifier.getDatabaseName(), identifier.getObjectName()); + return ObjectIdentifier.of( + identifier.getCatalogName(), identifier.getDatabaseName(), parsed.getTableName()); + } + + public static ObjectPath stripRealTimeSuffix(ObjectPath tablePath) { + return new ObjectPath( + tablePath.getDatabaseName(), FlinkCatalog.toIdentifier(tablePath).getTableName()); + } + + /** + * Extracts every entry whose key starts with {@code prefix}, with the prefix removed, e.g. + * {@code fluss.bootstrap.servers} becomes {@code bootstrap.servers} for the prefix {@code + * fluss.}. + */ + public static Map<String, String> extractPrefixedOptions( + Map<String, String> options, String prefix) { + Map<String, String> result = new HashMap<>(); + for (Map.Entry<String, String> entry : options.entrySet()) { + if (entry.getKey().startsWith(prefix)) { + result.put(entry.getKey().substring(prefix.length()), entry.getValue()); + } + } + return result; + } + + /** The persisted Paimon schema used for lake-stream compatibility checks. */ + public static Schema toPaimonSchema(Table table) { + return new Schema( + table.rowType().getFields(), + table.partitionKeys(), + table.primaryKeys(), + table.options(), + table.comment().orElse(null)); + } + + /** A store table expressed as a Paimon schema with only shared table options retained. */ + public static Schema toPaimonSchema(CatalogBaseTable table) { + if (!(table instanceof CatalogTable)) { + throw new ValidationException( + "Expected a table from the Lakestream Store, but got " + + table.getClass().getName()); + } + org.apache.flink.table.api.Schema storeSchema = table.getUnresolvedSchema(); + Schema.Builder builder = + Schema.newBuilder() + .partitionKeys(((CatalogTable) table).getPartitionKeys()) + .options(sharedPaimonOptions(table.getOptions())) + .comment(table.getComment()); + for (org.apache.flink.table.api.Schema.UnresolvedColumn column : storeSchema.getColumns()) { + if (column instanceof org.apache.flink.table.api.Schema.UnresolvedPhysicalColumn) { + builder.column( + column.getName(), + toPaimonType( + ((org.apache.flink.table.api.Schema.UnresolvedPhysicalColumn) + column) + .getDataType()), + column.getComment().orElse(null)); + } + } + storeSchema + .getPrimaryKey() + .ifPresent(primaryKey -> builder.primaryKey(primaryKey.getColumnNames())); + return builder.build(); + } + + /** Whether physical fields, keys and shared immutable options match. */ + public static boolean isCompatible(Schema paimonSchema, Schema storeSchema) { + return equalFields(paimonSchema.fields(), storeSchema.fields()) + && paimonSchema.partitionKeys().equals(storeSchema.partitionKeys()) + && paimonSchema.primaryKeys().equals(storeSchema.primaryKeys()) + && equalSelectedOptions(paimonSchema.options(), storeSchema.options()); + } + + private static org.apache.paimon.types.DataType toPaimonType(AbstractDataType<?> type) { + return LogicalTypeConversion.toDataType(((DataType) type).getLogicalType()); + } + + private static Map<String, String> sharedPaimonOptions(Map<String, String> options) { + Map<String, String> result = new HashMap<>(); + for (Map.Entry<String, String> entry : options.entrySet()) { + String key = entry.getKey(); + if (key.startsWith("paimon.")) { + key = key.substring("paimon.".length()); + } + if (isComparedOption(key)) { + result.put(key, entry.getValue()); + } + } + return result; + } + + private static boolean equalFields(List<DataField> paimonFields, List<DataField> storeFields) { + if (paimonFields.size() != storeFields.size()) { + return false; + } + for (int i = 0; i < paimonFields.size(); i++) { + DataField paimonField = paimonFields.get(i); + DataField storeField = storeFields.get(i); + if (!paimonField.name().equals(storeField.name()) + || !paimonField.type().equalsIgnoreFieldId(storeField.type())) { + return false; Review Comment: [P2] Ignore nested field comments when comparing physical layouts `equalsIgnoreFieldId` still compares nested DataField descriptions through ROW types. Against the real Fluss 1.0 cluster, both sides declared `r ROW<x INT>`; the first INSERT and real-time read succeeded despite differing top-level comments. I then changed only Paimon's `r.x` comment with `SchemaChange.updateColumnComment`: the next INSERT fails with 'have not caught up', although field names, order, types, nullability and keys are unchanged. This makes a metadata-only edit stop all subsequent writes. Compare the physical type recursively while ignoring nested comments, preserving the checks for names/order/types/nullability. Removing only that metadata difference makes the same write/read regression pass. ########## paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lakestream/LakeStreamTableFactory.java: ########## @@ -0,0 +1,365 @@ +/* + * 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.fs.FileIO; +import org.apache.paimon.fs.Path; +import org.apache.paimon.options.CatalogOptions; +import org.apache.paimon.options.Options; +import org.apache.paimon.schema.Schema; +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.io.IOException; +import java.io.UncheckedIOException; +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()); + checkSchemaAlignment(table, storeTable, identifier); + DynamicTableSinkFactory sinkFactory = + storeFactory( + storeCatalog, DynamicTableSinkFactory.class, "create a sink for writes"); + return new LakeStreamDynamicTableSink( + sinkFactory.createDynamicTableSink( + delegatedContext(context, identifier, storeTable)), Review Comment: [P2] Preserve statement-local provider options on the store sink The sink path fetches a fresh store table and delegates it without applying the SQL OPTIONS hints from the original context table. I executed `INSERT INTO t /*+ OPTIONS('sink.producer-id'='review-stable-producer') */ SELECT ...` with dynamic table options enabled: the row reaches the store, but its captured sink context contains no producer ID. Actual Fluss 1.0 reads this option from the table options and otherwise falls back to the job ID, so the requested stable identity across submissions is silently lost. Copying the global Configuration does not recover a table hint; `sink.ignore-delete` is affected too. Keep the store's base options and apply valid statement-local provider overrides, while retaining persisted metadata for routing and schema checks. The same SQL test passes when the producer hint is forwarded. -- 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]
