twalthr commented on a change in pull request #18283: URL: https://github.com/apache/flink/pull/18283#discussion_r779547250
########## File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaConnectorOptionsUtil.java ########## @@ -519,6 +519,7 @@ private static boolean hasKafkaClientProperties(Map<String, String> tableOptions return new FactoryUtil.DefaultDynamicTableContext( context.getObjectIdentifier(), context.getCatalogTable().copy(newOptions), Review comment: side note: what is this method doing? could it be a problem for the upgrade story? ########## File path: flink-table/flink-table-common/src/main/java/org/apache/flink/table/factories/FormatFactory.java ########## @@ -0,0 +1,33 @@ +/* + * 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.flink.table.factories; + +import org.apache.flink.configuration.ConfigOption; + +import java.util.Collections; +import java.util.Set; + +/** Base interface for {@link DecodingFormatFactory} and {@link EncodingFormatFactory}. */ +public interface FormatFactory extends Factory { Review comment: add annotation ########## File path: flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicTableFactory.java ########## @@ -148,7 +148,15 @@ public String factoryIdentifier() { public DynamicTableSource createDynamicTableSource(Context context) { final TableFactoryHelper helper = FactoryUtil.createTableFactoryHelper(this, context); - final ReadableConfig tableOptions = helper.getOptions(); + final ReadableConfig tableOptions = + helper.forwardOptions( + TOPIC, + TOPIC_PATTERN, + SCAN_STARTUP_MODE, + SCAN_STARTUP_SPECIFIC_OFFSETS, + SCAN_TOPIC_PARTITION_DISCOVERY, + SCAN_STARTUP_TIMESTAMP_MILLIS) + .getOptions(); Review comment: let's call getOptions() after calling `validate`. maybe we should not offer them as a return type of `forwardOptions`. ########## File path: flink-table/flink-table-common/src/main/java/org/apache/flink/table/factories/FactoryUtil.java ########## @@ -1035,8 +1124,62 @@ private String formatPrefix(Factory formatFactory, ConfigOption<String> formatOp return getFormatPrefix(formatOption, identifier); } - private ReadableConfig projectOptions(String formatPrefix) { - return new DelegatingConfiguration(allOptions, formatPrefix); + @SuppressWarnings({"unchecked"}) + private ReadableConfig createFormatOptions( + String formatPrefix, FormatFactory formatFactory) { + Set<ConfigOption<?>> mergeableConfigOptions = formatFactory.mergeableOptions(); + Configuration formatConf = new DelegatingConfiguration(allOptions, formatPrefix); + if (mergeableConfigOptions.isEmpty()) { + return formatConf; + } + + Configuration formatConfFromMergeableOptions = + new DelegatingConfiguration(mergeableOptions, formatPrefix); + + for (ConfigOption<?> option : mergeableConfigOptions) { + formatConfFromMergeableOptions + .getOptional(option) + .ifPresent(o -> formatConf.set((ConfigOption<? super Object>) option, o)); + } + + return formatConf; + } + + /** + * This function assumes that the format config is used only and only if the original + * configuration contains the format config option. It will fail if there is a mismatch of + * the identifier between the format in originalMap and the one in catalog. + */ + private void checkFormatIdentifierMatchesWithMergeableOptions( + ConfigOption<String> formatOption, + String resolvedIdentifierFromContextResolvedCatalogTable) { + Optional<String> identifierFromMergeableOptions = + mergeableOptions.getOptional(formatOption); + + if (!identifierFromMergeableOptions.isPresent()) { + return; + } + + if (resolvedIdentifierFromContextResolvedCatalogTable == null) { + throw new ValidationException( + String.format( + "The persisted plan has no format option '%s' specified, while the catalog table has it with value '%s'. " Review comment: also: ``` while the enriching catalog table has it with ``` ########## File path: flink-table/flink-table-common/src/main/java/org/apache/flink/table/factories/FactoryUtil.java ########## @@ -1035,8 +1124,62 @@ private String formatPrefix(Factory formatFactory, ConfigOption<String> formatOp return getFormatPrefix(formatOption, identifier); } - private ReadableConfig projectOptions(String formatPrefix) { - return new DelegatingConfiguration(allOptions, formatPrefix); + @SuppressWarnings({"unchecked"}) + private ReadableConfig createFormatOptions( + String formatPrefix, FormatFactory formatFactory) { + Set<ConfigOption<?>> mergeableConfigOptions = formatFactory.mergeableOptions(); + Configuration formatConf = new DelegatingConfiguration(allOptions, formatPrefix); + if (mergeableConfigOptions.isEmpty()) { + return formatConf; + } + + Configuration formatConfFromMergeableOptions = + new DelegatingConfiguration(mergeableOptions, formatPrefix); + + for (ConfigOption<?> option : mergeableConfigOptions) { + formatConfFromMergeableOptions + .getOptional(option) + .ifPresent(o -> formatConf.set((ConfigOption<? super Object>) option, o)); + } + + return formatConf; + } + + /** + * This function assumes that the format config is used only and only if the original + * configuration contains the format config option. It will fail if there is a mismatch of + * the identifier between the format in originalMap and the one in catalog. Review comment: what is `originalMap`? ########## File path: flink-table/flink-table-common/src/main/java/org/apache/flink/table/factories/FactoryUtil.java ########## @@ -1035,8 +1124,62 @@ private String formatPrefix(Factory formatFactory, ConfigOption<String> formatOp return getFormatPrefix(formatOption, identifier); } - private ReadableConfig projectOptions(String formatPrefix) { - return new DelegatingConfiguration(allOptions, formatPrefix); + @SuppressWarnings({"unchecked"}) + private ReadableConfig createFormatOptions( + String formatPrefix, FormatFactory formatFactory) { + Set<ConfigOption<?>> mergeableConfigOptions = formatFactory.mergeableOptions(); + Configuration formatConf = new DelegatingConfiguration(allOptions, formatPrefix); + if (mergeableConfigOptions.isEmpty()) { + return formatConf; + } + + Configuration formatConfFromMergeableOptions = + new DelegatingConfiguration(mergeableOptions, formatPrefix); + + for (ConfigOption<?> option : mergeableConfigOptions) { + formatConfFromMergeableOptions + .getOptional(option) + .ifPresent(o -> formatConf.set((ConfigOption<? super Object>) option, o)); + } + + return formatConf; + } + + /** + * This function assumes that the format config is used only and only if the original + * configuration contains the format config option. It will fail if there is a mismatch of + * the identifier between the format in originalMap and the one in catalog. + */ + private void checkFormatIdentifierMatchesWithMergeableOptions( + ConfigOption<String> formatOption, + String resolvedIdentifierFromContextResolvedCatalogTable) { + Optional<String> identifierFromMergeableOptions = + mergeableOptions.getOptional(formatOption); + + if (!identifierFromMergeableOptions.isPresent()) { + return; + } + + if (resolvedIdentifierFromContextResolvedCatalogTable == null) { + throw new ValidationException( + String.format( + "The persisted plan has no format option '%s' specified, while the catalog table has it with value '%s'. " + + "This is invalid, as either only the persisted plan table defines the format, " + + "or both the persisted plan table and the catalog table defines the same format", Review comment: nit: dot at the end ########## File path: flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonFormatFactory.java ########## @@ -151,4 +151,15 @@ public String factoryIdentifier() { options.add(ENCODE_DECIMAL_AS_PLAIN_NUMBER); return options; } + + @Override + public Set<ConfigOption<?>> mergeableOptions() { Review comment: Call this `forwardOptions` similar to helper? Because those are not only mergeable but will be forwarded. ########## File path: flink-table/flink-table-common/src/main/java/org/apache/flink/table/factories/FactoryUtil.java ########## @@ -1035,8 +1124,62 @@ private String formatPrefix(Factory formatFactory, ConfigOption<String> formatOp return getFormatPrefix(formatOption, identifier); } - private ReadableConfig projectOptions(String formatPrefix) { - return new DelegatingConfiguration(allOptions, formatPrefix); + @SuppressWarnings({"unchecked"}) + private ReadableConfig createFormatOptions( + String formatPrefix, FormatFactory formatFactory) { + Set<ConfigOption<?>> mergeableConfigOptions = formatFactory.mergeableOptions(); + Configuration formatConf = new DelegatingConfiguration(allOptions, formatPrefix); + if (mergeableConfigOptions.isEmpty()) { + return formatConf; + } + + Configuration formatConfFromMergeableOptions = + new DelegatingConfiguration(mergeableOptions, formatPrefix); + + for (ConfigOption<?> option : mergeableConfigOptions) { + formatConfFromMergeableOptions + .getOptional(option) + .ifPresent(o -> formatConf.set((ConfigOption<? super Object>) option, o)); + } + + return formatConf; + } + + /** + * This function assumes that the format config is used only and only if the original + * configuration contains the format config option. It will fail if there is a mismatch of + * the identifier between the format in originalMap and the one in catalog. + */ + private void checkFormatIdentifierMatchesWithMergeableOptions( + ConfigOption<String> formatOption, + String resolvedIdentifierFromContextResolvedCatalogTable) { + Optional<String> identifierFromMergeableOptions = + mergeableOptions.getOptional(formatOption); + + if (!identifierFromMergeableOptions.isPresent()) { + return; + } + + if (resolvedIdentifierFromContextResolvedCatalogTable == null) { + throw new ValidationException( + String.format( + "The persisted plan has no format option '%s' specified, while the catalog table has it with value '%s'. " + + "This is invalid, as either only the persisted plan table defines the format, " + + "or both the persisted plan table and the catalog table defines the same format", Review comment: also: ``` and enriching catalog table defines... ``` ########## File path: flink-table/flink-table-common/src/main/java/org/apache/flink/table/factories/FactoryUtil.java ########## @@ -1035,8 +1124,62 @@ private String formatPrefix(Factory formatFactory, ConfigOption<String> formatOp return getFormatPrefix(formatOption, identifier); } - private ReadableConfig projectOptions(String formatPrefix) { - return new DelegatingConfiguration(allOptions, formatPrefix); + @SuppressWarnings({"unchecked"}) + private ReadableConfig createFormatOptions( + String formatPrefix, FormatFactory formatFactory) { + Set<ConfigOption<?>> mergeableConfigOptions = formatFactory.mergeableOptions(); + Configuration formatConf = new DelegatingConfiguration(allOptions, formatPrefix); + if (mergeableConfigOptions.isEmpty()) { + return formatConf; + } + + Configuration formatConfFromMergeableOptions = + new DelegatingConfiguration(mergeableOptions, formatPrefix); + + for (ConfigOption<?> option : mergeableConfigOptions) { + formatConfFromMergeableOptions + .getOptional(option) + .ifPresent(o -> formatConf.set((ConfigOption<? super Object>) option, o)); + } + + return formatConf; + } + + /** + * This function assumes that the format config is used only and only if the original + * configuration contains the format config option. It will fail if there is a mismatch of + * the identifier between the format in originalMap and the one in catalog. + */ + private void checkFormatIdentifierMatchesWithMergeableOptions( + ConfigOption<String> formatOption, + String resolvedIdentifierFromContextResolvedCatalogTable) { + Optional<String> identifierFromMergeableOptions = + mergeableOptions.getOptional(formatOption); + + if (!identifierFromMergeableOptions.isPresent()) { + return; + } + + if (resolvedIdentifierFromContextResolvedCatalogTable == null) { + throw new ValidationException( + String.format( + "The persisted plan has no format option '%s' specified, while the catalog table has it with value '%s'. " Review comment: Let's synchronize the variable names with the exception. `resolvedIdentifierFromContextResolvedCatalogTable` -> `identifierFromPlan`? -- 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: issues-unsubscr...@flink.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org