[FLINK-17633][table-common] Improve FactoryUtil to align with new format option keys

This closes #12099

flink-table/flink-table-common/src/main/java/org/apache/flink/table/factories/FactoryUtil.java

@@ -71,11 +71,30 @@ public final class FactoryUtil {
 			"Uniquely identifies the connector of a dynamic table that is used for accessing data in " +
 			"an external system. Its value is used during table source and table sink discovery.");
 
-	public static final String FORMAT_PREFIX = "format.";
+	public static final ConfigOption<String> KEY_FORMAT = ConfigOptions
+		.key("key.format")
+		.stringType()
+		.noDefaultValue()
+		.withDescription("Defines the format identifier for encoding key data. " +
+			"The identifier is used to discover a suitable format factory.");
 
-	public static final String KEY_FORMAT_PREFIX = "key.format.";
+	public static final ConfigOption<String> VALUE_FORMAT = ConfigOptions
+		.key("value.format")
+		.stringType()
+		.noDefaultValue()
+		.withDescription("Defines the format identifier for encoding value data. " +
+			"The identifier is used to discover a suitable format factory.");
+
+	public static final ConfigOption<String> FORMAT = ConfigOptions
+		.key("format")
+		.stringType()
+		.noDefaultValue()
+		.withDescription("Defines the format identifier for encoding data. " +
+			"The identifier is used to discover a suitable format factory.");
 
-	public static final String VALUE_FORMAT_PREFIX = "value.format.";
+	private static final String FORMAT_KEY = "format";
+
+	private static final String FORMAT_SUFFIX = ".format";
 
 	/**
 	 * Creates a {@link DynamicTableSource} from a {@link CatalogTable}.
@@ -162,12 +181,20 @@ public final class FactoryUtil {
 	 * <pre>{@code
 	 * // in createDynamicTableSource()
 	 * helper = FactoryUtil.createTableFactoryHelper(this, context);
-	 * keyFormat = helper.discoverScanFormat(classloader, MyFormatFactory.class, KEY_OPTION, "prefix");
-	 * valueFormat = helper.discoverScanFormat(classloader, MyFormatFactory.class, VALUE_OPTION, "prefix");
+	 * keyFormat = helper.discoverScanFormat(DeserializationFormatFactory.class, KEY_FORMAT);
+	 * valueFormat = helper.discoverScanFormat(DeserializationFormatFactory.class, VALUE_FORMAT);
 	 * helper.validate();
 	 * ... // construct connector with discovered formats
 	 * }</pre>
 	 *
+	 * <p>Note: The format option parameter of {@code helper.discoverScanFormat(formatFactoryClass, formatOption)}
+	 * and {@code helper.discoverSinkFormat(formatFactoryClass, formatOption)} must be 'format' or
+	 * with '.format' suffix (e.g. {@link #FORMAT}, {@link #KEY_FORMAT} and {@link #VALUE_FORMAT}).
+	 * The discovery logic will replace 'format' with the factory identifier value as the format
+	 * prefix. For example, assuming the identifier is 'json', if format option key is 'format',
+	 * then format prefix is 'json.'. If format option key is 'value.format', then format prefix
+	 * is 'value.json'. The format prefix is used to project the options for the format factory.
+	 *
 	 * <p>Note: This utility checks for left-over options in the final step.
 	 */
 	public static TableFactoryHelper createTableFactoryHelper(
@@ -377,14 +404,11 @@ public final class FactoryUtil {
 
 		/**
 		 * Discovers a {@link ScanFormat} of the given type using the given option as factory identifier.
-		 *
-		 * <p>A prefix, e.g. {@link #KEY_FORMAT_PREFIX}, projects the options for the format factory.
 		 */
 		public <I, F extends ScanFormatFactory<I>> ScanFormat<I> discoverScanFormat(
 				Class<F> formatFactoryClass,
-				ConfigOption<String> formatOption,
-				String formatPrefix) {
-			return discoverOptionalScanFormat(formatFactoryClass, formatOption, formatPrefix)
+				ConfigOption<String> formatOption) {
+			return discoverOptionalScanFormat(formatFactoryClass, formatOption)
 				.orElseThrow(() ->
 					new ValidationException(
 						String.format("Could not find required scan format '%s'.", formatOption.key())));
@@ -393,15 +417,13 @@ public final class FactoryUtil {
 		/**
 		 * Discovers a {@link ScanFormat} of the given type using the given option (if present) as factory
 		 * identifier.
-		 *
-		 * <p>A prefix, e.g. {@link #KEY_FORMAT_PREFIX}, projects the options for the format factory.
 		 */
 		public <I, F extends ScanFormatFactory<I>> Optional<ScanFormat<I>> discoverOptionalScanFormat(
 				Class<F> formatFactoryClass,
-				ConfigOption<String> formatOption,
-				String formatPrefix) {
-			return discoverOptionalFormatFactory(formatFactoryClass, formatOption, formatPrefix)
+				ConfigOption<String> formatOption) {
+			return discoverOptionalFormatFactory(formatFactoryClass, formatOption)
 				.map(formatFactory -> {
+					String formatPrefix = formatPrefix(formatFactory, formatOption);
 					try {
 						return formatFactory.createScanFormat(context, projectOptions(formatPrefix));
 					} catch (Throwable t) {
@@ -417,14 +439,11 @@ public final class FactoryUtil {
 
 		/**
 		 * Discovers a {@link SinkFormat} of the given type using the given option as factory identifier.
-		 *
-		 * <p>A prefix, e.g. {@link #KEY_FORMAT_PREFIX}, projects the options for the format factory.
 		 */
 		public <I, F extends SinkFormatFactory<I>> SinkFormat<I> discoverSinkFormat(
 				Class<F> formatFactoryClass,
-				ConfigOption<String> formatOption,
-				String formatPrefix) {
-			return discoverOptionalSinkFormat(formatFactoryClass, formatOption, formatPrefix)
+				ConfigOption<String> formatOption) {
+			return discoverOptionalSinkFormat(formatFactoryClass, formatOption)
 				.orElseThrow(() ->
 					new ValidationException(
 						String.format("Could not find required sink format '%s'.", formatOption.key())));
@@ -433,15 +452,13 @@ public final class FactoryUtil {
 		/**
 		 * Discovers a {@link SinkFormat} of the given type using the given option (if present) as factory
 		 * identifier.
-		 *
-		 * <p>A prefix, e.g. {@link #KEY_FORMAT_PREFIX}, projects the options for the format factory.
 		 */
 		public <I, F extends SinkFormatFactory<I>> Optional<SinkFormat<I>> discoverOptionalSinkFormat(
 				Class<F> formatFactoryClass,
-				ConfigOption<String> formatOption,
-				String formatPrefix) {
-			return discoverOptionalFormatFactory(formatFactoryClass, formatOption, formatPrefix)
+				ConfigOption<String> formatOption) {
+			return discoverOptionalFormatFactory(formatFactoryClass, formatOption)
 				.map(formatFactory -> {
+					String formatPrefix = formatPrefix(formatFactory, formatOption);
 					try {
 						return formatFactory.createSinkFormat(context, projectOptions(formatPrefix));
 					} catch (Throwable t) {
@@ -492,8 +509,7 @@ public final class FactoryUtil {
 
 		private <F extends Factory> Optional<F> discoverOptionalFormatFactory(
 				Class<F> formatFactoryClass,
-				ConfigOption<String> formatOption,
-				String formatPrefix) {
+				ConfigOption<String> formatOption) {
 			final String identifier = allOptions.get(formatOption);
 			if (identifier == null) {
 				return Optional.empty();
@@ -502,6 +518,7 @@ public final class FactoryUtil {
 				context.getClassLoader(),
 				formatFactoryClass,
 				identifier);
+			String formatPrefix = formatPrefix(factory, formatOption);
 			// log all used options of other factories
 			consumedOptionKeys.addAll(
 				factory.requiredOptions().stream()
@@ -516,6 +533,21 @@ public final class FactoryUtil {
 			return Optional.of(factory);
 		}
 
+		private String formatPrefix(Factory formatFactory, ConfigOption<String> formatOption) {
+			String identifier = formatFactory.factoryIdentifier();
+			if (formatOption.key().equals(FORMAT_KEY)) {
+				return identifier + ".";
+			} else if (formatOption.key().endsWith(FORMAT_SUFFIX)) {
+				// extract the key prefix, e.g. extract 'key' from 'key.format'
+				String keyPrefix = formatOption.key().substring(0, formatOption.key().length() - FORMAT_SUFFIX.length());
+				return keyPrefix + "." + identifier + ".";
+			} else {
+				throw new ValidationException(
+					"Format identifier key should be 'format' or suffix with '.format', " +
+						"don't support format identifier key '" + formatOption.key() + "'.");
+			}
+		}
+
 		private ReadableConfig projectOptions(String formatPrefix) {
 			return new DelegatingConfiguration(
 				allOptions,

flink-table/flink-table-common/src/test/java/org/apache/flink/table/factories/FactoryUtilTest.java

@@ -85,8 +85,8 @@ public class FactoryUtilTest {
 
 	@Test
 	public void testMissingFormat() {
-		expectError("Could not find required scan format 'value.format.kind'.");
-		testError(options -> options.remove("value.format.kind"));
+		expectError("Could not find required scan format 'value.format'.");
+		testError(options -> options.remove("value.format"));
 	}
 
 	@Test
@@ -96,24 +96,24 @@ public class FactoryUtilTest {
 				DeserializationFormatFactory.class.getName() + "' in the classpath.\n\n" +
 			"Available factory identifiers are:\n\n" +
 			"test-format");
-		testError(options -> options.put("value.format.kind", "FAIL"));
+		testError(options -> options.put("value.format", "FAIL"));
 	}
 
 	@Test
 	public void testMissingFormatOption() {
 		expectError(
-			"Error creating scan format 'test-format' in option space 'key.format.'.");
+			"Error creating scan format 'test-format' in option space 'key.test-format.'.");
 		expectError(
 			"One or more required options are missing.\n\n" +
 			"Missing required options are:\n\n" +
 			"delimiter");
-		testError(options -> options.remove("key.format.delimiter"));
+		testError(options -> options.remove("key.test-format.delimiter"));
 	}
 
 	@Test
 	public void testInvalidFormatOption() {
 		expectError("Invalid value for option 'fail-on-missing'.");
-		testError(options -> options.put("key.format.fail-on-missing", "FAIL"));
+		testError(options -> options.put("key.test-format.fail-on-missing", "FAIL"));
 	}
 
 	@Test
@@ -126,14 +126,15 @@ public class FactoryUtilTest {
 			"Supported options:\n\n" +
 			"buffer-size\n" +
 			"connector\n" +
-			"key.format.delimiter\n" +
-			"key.format.fail-on-missing\n" +
-			"key.format.kind\n" +
+			"format\n" +
+			"key.format\n" +
+			"key.test-format.delimiter\n" +
+			"key.test-format.fail-on-missing\n" +
 			"property-version\n" +
 			"target\n" +
-			"value.format.delimiter\n" +
-			"value.format.fail-on-missing\n" +
-			"value.format.kind");
+			"value.format\n" +
+			"value.test-format.delimiter\n" +
+			"value.test-format.fail-on-missing");
 		testError(options -> {
 			options.put("this-is-not-consumed", "42");
 			options.put("this-is-also-not-consumed", "true");
@@ -161,8 +162,8 @@ public class FactoryUtilTest {
 	@Test
 	public void testOptionalFormat() {
 		final Map<String, String> options = createAllOptions();
-		options.remove("key.format.kind");
-		options.remove("key.format.delimiter");
+		options.remove("key.format");
+		options.remove("key.test-format.delimiter");
 		final DynamicTableSource actualSource = createTableSource(options);
 		final DynamicTableSource expectedSource = new DynamicTableSourceMock(
 			"MyTarget",
@@ -178,6 +179,30 @@ public class FactoryUtilTest {
 		assertEquals(expectedSink, actualSink);
 	}
 
+	@Test
+	public void testAlternativeValueFormat() {
+		final Map<String, String> options = createAllOptions();
+		options.remove("value.format");
+		options.remove("value.test-format.delimiter");
+		options.remove("value.test-format.fail-on-missing");
+		options.put("format", "test-format");
+		options.put("test-format.delimiter", ";");
+		options.put("test-format.fail-on-missing", "true");
+		final DynamicTableSource actualSource = createTableSource(options);
+		final DynamicTableSource expectedSource = new DynamicTableSourceMock(
+			"MyTarget",
+			new ScanFormatMock(",", false),
+			new ScanFormatMock(";", true));
+		assertEquals(expectedSource, actualSource);
+		final DynamicTableSink actualSink = createTableSink(options);
+		final DynamicTableSink expectedSink = new DynamicTableSinkMock(
+			"MyTarget",
+			1000L,
+			new SinkFormatMock(","),
+			new SinkFormatMock(";"));
+		assertEquals(expectedSink, actualSink);
+	}
+
 	// --------------------------------------------------------------------------------------------
 
 	private void expectError(String message) {
@@ -198,11 +223,11 @@ public class FactoryUtilTest {
 		options.put("connector", TestDynamicTableFactory.IDENTIFIER);
 		options.put("target", "MyTarget");
 		options.put("buffer-size", "1000");
-		options.put("key.format.kind", TestFormatFactory.IDENTIFIER);
-		options.put("key.format.delimiter", ",");
-		options.put("value.format.kind", TestFormatFactory.IDENTIFIER);
-		options.put("value.format.delimiter", "|");
-		options.put("value.format.fail-on-missing", "true");
+		options.put("key.format", "test-format");
+		options.put("key.test-format.delimiter", ",");
+		options.put("value.format", "test-format");
+		options.put("value.test-format.delimiter", "|");
+		options.put("value.test-format.fail-on-missing", "true");
 		return options;
 	}
 

flink-table/flink-table-common/src/test/java/org/apache/flink/table/factories/TestDynamicTableFactory.java

@@ -38,6 +38,10 @@ import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
 
+import static org.apache.flink.table.factories.FactoryUtil.FORMAT;
+import static org.apache.flink.table.factories.FactoryUtil.KEY_FORMAT;
+import static org.apache.flink.table.factories.FactoryUtil.VALUE_FORMAT;
+
 /**
  * Test implementations for {@link DynamicTableSourceFactory} and {@link DynamicTableSinkFactory}.
  */
@@ -55,28 +59,19 @@ public final class TestDynamicTableFactory implements DynamicTableSourceFactory,
 		.longType()
 		.defaultValue(100L);
 
-	public static final ConfigOption<String> KEY_FORMAT = ConfigOptions
-		.key("key.format.kind")
-		.stringType()
-		.noDefaultValue();
-
-	public static final ConfigOption<String> VALUE_FORMAT = ConfigOptions
-		.key("value.format.kind")
-		.stringType()
-		.noDefaultValue();
-
 	@Override
 	public DynamicTableSource createDynamicTableSource(Context context) {
 		final TableFactoryHelper helper = FactoryUtil.createTableFactoryHelper(this, context);
 
 		final Optional<ScanFormat<DeserializationSchema<RowData>>> keyFormat = helper.discoverOptionalScanFormat(
 			DeserializationFormatFactory.class,
-			KEY_FORMAT,
-			FactoryUtil.KEY_FORMAT_PREFIX);
-		final ScanFormat<DeserializationSchema<RowData>> valueFormat = helper.discoverScanFormat(
+			KEY_FORMAT);
+		final ScanFormat<DeserializationSchema<RowData>> valueFormat = helper.discoverOptionalScanFormat(
 			DeserializationFormatFactory.class,
-			VALUE_FORMAT,
-			FactoryUtil.VALUE_FORMAT_PREFIX);
+			FORMAT).orElseGet(
+				() -> helper.discoverScanFormat(
+					DeserializationFormatFactory.class,
+					VALUE_FORMAT));
 		helper.validate();
 
 		return new DynamicTableSourceMock(
@@ -91,12 +86,13 @@ public final class TestDynamicTableFactory implements DynamicTableSourceFactory,
 
 		final Optional<SinkFormat<SerializationSchema<RowData>>> keyFormat = helper.discoverOptionalSinkFormat(
 			SerializationFormatFactory.class,
-			KEY_FORMAT,
-			FactoryUtil.KEY_FORMAT_PREFIX);
-		final SinkFormat<SerializationSchema<RowData>> valueFormat = helper.discoverSinkFormat(
+			KEY_FORMAT);
+		final SinkFormat<SerializationSchema<RowData>> valueFormat = helper.discoverOptionalSinkFormat(
 			SerializationFormatFactory.class,
-			VALUE_FORMAT,
-			FactoryUtil.VALUE_FORMAT_PREFIX);
+			FORMAT).orElseGet(
+				() -> helper.discoverSinkFormat(
+					SerializationFormatFactory.class,
+					VALUE_FORMAT));
 		helper.validate();
 
 		return new DynamicTableSinkMock(
@@ -115,7 +111,6 @@ public final class TestDynamicTableFactory implements DynamicTableSourceFactory,
 	public Set<ConfigOption<?>> requiredOptions() {
 		final Set<ConfigOption<?>> options = new HashSet<>();
 		options.add(TARGET);
-		options.add(VALUE_FORMAT);
 		return options;
 	}
 
@@ -124,6 +119,8 @@ public final class TestDynamicTableFactory implements DynamicTableSourceFactory,
 		final Set<ConfigOption<?>> options = new HashSet<>();
 		options.add(BUFFER_SIZE);
 		options.add(KEY_FORMAT);
+		options.add(FORMAT);
+		options.add(VALUE_FORMAT);
 		return options;
 	}