You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by twalthr <gi...@git.apache.org> on 2018/07/05 10:48:53 UTC
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
GitHub user twalthr opened a pull request:
https://github.com/apache/flink/pull/6264
[FLINK-8558] [table] Add unified format interfaces and separate formats from connectors
## What is the purpose of the change
This PR introduces a format discovery mechanism based on Java Service Providers. The general `TableFormatFactory` is similar to the existing table source discovery mechanism. However, it allows for arbitrary format interfaces that might be introduced in the future. At the moment, a connector can request configured instances of `DeserializationSchema` and `SerializationSchema`. In the future we can add interfaces such as a `Writer` or `KeyedSerializationSchema` without breaking backwards compatibility.
This PR deprecates the existing strong coupling of connector and format for the Kafa table sources and table source factories. It introduces descriptor-based alternatives.
## Brief change log
- Introduction of `TableFormatService` with `TableFormatFactory` and specific `DeserializationSchemaFactory` and `SerializationSchemaFactory`
- Decoupling of existing connectors (i.e. Kafka) from formats (i.e. JSON and Avro)
- Exposing the descriptor-based approach, deprecate the old builders and make table source internal
## Verifying this change
- Existing tests for coupled sources and factories are still working
- New tests for format discovery, formats, and decoupled table sources
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: yes
- The serializers: no
- The runtime per-record code paths (performance sensitive): no
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: no
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? yes
- If yes, how is the feature documented? more docs for the descriptors will follow in a separate PR
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/twalthr/flink FLINK-8558
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/6264.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #6264
----
commit 9415066578d6fac420ea68edfa4509d5625f2a82
Author: Timo Walther <tw...@...>
Date: 2018-06-27T11:16:49Z
[FLINK-8558] [table] Add unified format interfaces and separate formats from connectors
This PR introduces a format discovery mechanism based on Java Service Providers. The general `TableFormatFactory` is similar to the existing table source discovery mechanism. However, it allows for arbirary format interfaces that might be introduced in the future. At the moment, a connector can request configured instances of `DeserializationSchema` and `SerializationSchema`. In the future we can add interfaces such as a `Writer` or `KeyedSerializationSchema` without breaking backwards compatibility.
This PR deprecates the existing strong coupling of connector and format for the Kafa table sources and table source factories. It introduces descriptor-based alternatives.
----
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200597211
--- Diff: flink-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010JsonTableSource.java ---
@@ -32,7 +33,13 @@
/**
* Kafka {@link StreamTableSource} for Kafka 0.10.
+ *
+ * @deprecated Use the {@link org.apache.flink.table.descriptors.Kafka} descriptor together
+ * with descriptors for schema and format instead. Descriptors allow for
+ * implementation-agnostic definition of tables. See also
+ * {@link org.apache.flink.table.api.TableEnvironment#from(ConnectorDescriptor)}.
*/
+@Deprecated
--- End diff --
I have mixed feelings about `Deprecating` `PublicEvolving` classes.
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200682089
--- Diff: flink-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010JsonTableSource.java ---
@@ -32,7 +33,13 @@
/**
* Kafka {@link StreamTableSource} for Kafka 0.10.
+ *
+ * @deprecated Use the {@link org.apache.flink.table.descriptors.Kafka} descriptor together
+ * with descriptors for schema and format instead. Descriptors allow for
+ * implementation-agnostic definition of tables. See also
+ * {@link org.apache.flink.table.api.TableEnvironment#from(ConnectorDescriptor)}.
*/
+@Deprecated
--- End diff --
I agree that having both annotation looks weird. I will remove the `PublicEvolving`.
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200908163
--- Diff: flink-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010TableSource.java ---
@@ -18,48 +18,99 @@
package org.apache.flink.streaming.connectors.kafka;
-import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.streaming.connectors.kafka.config.StartupMode;
+import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.sources.RowtimeAttributeDescriptor;
import org.apache.flink.table.sources.StreamTableSource;
import org.apache.flink.types.Row;
+import java.util.List;
+import java.util.Map;
import java.util.Properties;
/**
* Kafka {@link StreamTableSource} for Kafka 0.10.
*/
-@PublicEvolving
-public abstract class Kafka010TableSource extends KafkaTableSource {
+@Internal
+public class Kafka010TableSource extends KafkaTableSource {
- // The deserialization schema for the Kafka records
- private final DeserializationSchema<Row> deserializationSchema;
+ /**
+ * Creates a Kafka 0.10 {@link StreamTableSource}.
+ *
+ * @param schema Schema of the produced table.
+ * @param proctimeAttribute Field name of the processing time attribute, null if no
--- End diff --
https://stackoverflow.com/questions/31922866/why-should-java-8s-optional-not-be-used-in-arguments
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200687537
--- Diff: flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka011JsonTableSourceFactory.java ---
@@ -1,37 +0,0 @@
-/*
- * 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.streaming.connectors.kafka;
-
-import static org.apache.flink.table.descriptors.KafkaValidator.CONNECTOR_VERSION_VALUE_011;
-
-/**
- * Factory for creating configured instances of {@link Kafka011JsonTableSource}.
- */
-public class Kafka011JsonTableSourceFactory extends KafkaJsonTableSourceFactory {
--- End diff --
The factories were never public API and have been introduced in the last release. The sources were publicly documented:
https://ci.apache.org/projects/flink/flink-docs-master/dev/table/sourceSinks.html#provided-tablesources
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r201003278
--- Diff: flink-libraries/flink-table/src/test/scala/org/apache/flink/table/formats/TableFormatFactoryServiceTest.scala ---
@@ -0,0 +1,122 @@
+/*
+ * 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.formats
+
+import java.util.{HashMap => JHashMap, Map => JMap}
+
+import org.apache.flink.table.api.{AmbiguousTableFormatException, NoMatchingTableFormatException}
+import org.apache.flink.table.formats.utils.TestAmbiguousTableFormatFactory
+import org.junit.Assert.{assertNotNull, assertTrue}
+import org.junit.Test
+
+/**
+ * Tests for [[TableFormatFactoryService]].
+ */
+class TableFormatFactoryServiceTest {
+
+ @Test
+ def testValidProperties(): Unit = {
+ val props = properties()
+ assertNotNull(
--- End diff --
can `TableFormatFactoryService.find` return null?
If so, please change it to `Option` or `Optional` (or whatever scala equivalent is :()
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200990785
--- Diff: flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroRowFormatFactory.java ---
@@ -0,0 +1,97 @@
+/*
+ * 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.formats.avro;
+
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.table.descriptors.AvroValidator;
+import org.apache.flink.table.descriptors.DescriptorProperties;
+import org.apache.flink.table.descriptors.FormatDescriptorValidator;
+import org.apache.flink.table.formats.DeserializationSchemaFactory;
+import org.apache.flink.table.formats.SerializationSchemaFactory;
+import org.apache.flink.types.Row;
+
+import org.apache.avro.specific.SpecificRecord;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Table format factory for providing configured instances of Avro-to-row {@link SerializationSchema}
+ * and {@link DeserializationSchema}.
+ */
+public class AvroRowFormatFactory implements SerializationSchemaFactory<Row>, DeserializationSchemaFactory<Row> {
+
+ @Override
+ public Map<String, String> requiredContext() {
+ final Map<String, String> context = new HashMap<>();
+ context.put(FormatDescriptorValidator.FORMAT_TYPE(), AvroValidator.FORMAT_TYPE_VALUE);
+ context.put(FormatDescriptorValidator.FORMAT_PROPERTY_VERSION(), "1");
+ return context;
+ }
+
+ @Override
+ public boolean supportsSchemaDerivation() {
+ return false;
+ }
+
+ @Override
+ public List<String> supportedProperties() {
+ final List<String> properties = new ArrayList<>();
+ properties.add(AvroValidator.FORMAT_RECORD_CLASS);
+ properties.add(AvroValidator.FORMAT_AVRO_SCHEMA);
+ return properties;
+ }
+
+ @Override
+ public DeserializationSchema<Row> createDeserializationSchema(Map<String, String> properties) {
+ final DescriptorProperties props = new DescriptorProperties(true);
--- End diff --
I meant to extract following lines:
```
+ final DescriptorProperties props = new DescriptorProperties(true);
+ props.putProperties(properties);
+
+ // validate
+ new AvroValidator().validate(props);
```
to a separate private method and use it both in `createDeserializationSchema` and `createSerializationSchema`
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200987026
--- Diff: flink-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010TableSource.java ---
@@ -18,48 +18,99 @@
package org.apache.flink.streaming.connectors.kafka;
-import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.streaming.connectors.kafka.config.StartupMode;
+import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.sources.RowtimeAttributeDescriptor;
import org.apache.flink.table.sources.StreamTableSource;
import org.apache.flink.types.Row;
+import java.util.List;
+import java.util.Map;
import java.util.Properties;
/**
* Kafka {@link StreamTableSource} for Kafka 0.10.
*/
-@PublicEvolving
-public abstract class Kafka010TableSource extends KafkaTableSource {
+@Internal
+public class Kafka010TableSource extends KafkaTableSource {
- // The deserialization schema for the Kafka records
- private final DeserializationSchema<Row> deserializationSchema;
+ /**
+ * Creates a Kafka 0.10 {@link StreamTableSource}.
+ *
+ * @param schema Schema of the produced table.
+ * @param proctimeAttribute Field name of the processing time attribute, null if no
--- End diff --
This is kind of controversial topic. Generally speaking I suspect that Java discourage to use `Optional` beside return values because we should use `@Nullable` or not use any of them. However in projects that ignored `@Nullable` annotation (such as Flink), it's virtually impossible to start using them and thus using `Optional` is the only way to have a compiler control over optional/nullable fields.
In this particular use case of "optional" arguments my preference hierarchy is:
1. provide a builder for this class
2. provide alternative constructor without this argument
3. use `@Nullable` with enabled compile errors on incorrectly handled `@Nullable` annotations
4. use `Optional`
...
1336. use `@Nullable` WITHOUT compile errors on incorrectly handled `@Nullable` annotations
1337. use nullable argument without `@Nullable` annotation
two last options are for me out of the question, since `1337` is evil and `1336` doesn't improve situation. Option `3` is sadly impossible for Flink.
The same logic applies for me to other use cases (like fields, return values etc):
1. avoid nulls/optionals (for example via builders or named parameters with default values)
2. use `@Nullable` with compiler errors
3. use `Optional`
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200999523
--- Diff: flink-libraries/flink-table/src/test/scala/org/apache/flink/table/descriptors/TableSourceDescriptorTest.scala ---
@@ -0,0 +1,133 @@
+/*
+ * 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.descriptors
+
+import org.apache.flink.table.api.Types
+import org.apache.flink.table.utils.TableTestBase
+import org.junit.Assert.assertEquals
+import org.junit.Test
+
+import scala.collection.JavaConverters._
+
+class TableSourceDescriptorTest extends TableTestBase {
+
+ @Test
+ def testStreamTableSourceDescriptor(): Unit = {
+ val util = streamTestUtil()
+ val desc = util.tableEnv
--- End diff --
Not typed `desc` doesn't tell me what this field is (I even can not suggest how to rename it).
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200988428
--- Diff: flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSource.java ---
@@ -55,50 +57,101 @@
*/
@Internal
public abstract class KafkaTableSource
- implements StreamTableSource<Row>, DefinedProctimeAttribute, DefinedRowtimeAttributes {
+ implements StreamTableSource<Row>, DefinedProctimeAttribute, DefinedRowtimeAttributes, DefinedFieldMapping {
+
+ // common table source attributes
+ // TODO make all attributes final once we drop support for format-specific table sources
/** The schema of the table. */
private final TableSchema schema;
+ /** Field name of the processing time attribute, null if no processing time field is defined. */
+ private String proctimeAttribute;
+
+ /** Descriptor for a rowtime attribute. */
+ private List<RowtimeAttributeDescriptor> rowtimeAttributeDescriptors;
+
+ /** Mapping for the fields of the table schema to fields of the physical returned type or null. */
+ private Map<String, String> fieldMapping;
--- End diff --
please check my other comment about that. `@Nullable` without compiler errors is not in any way better :(
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200668536
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/descriptors/DescriptorProperties.scala ---
@@ -858,13 +859,18 @@ class DescriptorProperties(normalizeKeys: Boolean = true) {
/**
* Validates a type property.
*/
- def validateType(key: String, isOptional: Boolean): Unit = {
+ def validateType(key: String, requireRow: Boolean, isOptional: Boolean): Unit = {
if (!properties.contains(key)) {
if (!isOptional) {
throw new ValidationException(s"Could not find required property '$key'.")
}
} else {
- TypeStringUtils.readTypeInfo(properties(key)) // throws validation exceptions
+ // throws validation exceptions
--- End diff --
what's the purpose of this comment?
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200684172
--- Diff: flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceFactory.java ---
@@ -124,110 +123,130 @@
params.putProperties(properties);
// validate
- new SchemaValidator(true).validate(params);
+ // allow Kafka timestamps to be used, watermarks can not be received from source
+ new SchemaValidator(true, supportsKafkaTimestamps(), false).validate(params);
new KafkaValidator().validate(params);
- formatValidator().validate(params);
- // build
- final KafkaTableSource.Builder builder = createBuilderWithFormat(params);
--- End diff --
We don't need to maintain a builder for an internal class in the future. The table sources should be immutable classes. If we add a new parameter then it should be added to the constructor so that all invocations need to be updated.
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200670109
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/formats/TableFormatFactoryService.scala ---
@@ -0,0 +1,245 @@
+/*
+ * 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.formats
+
+import java.util.{ServiceConfigurationError, ServiceLoader, Map => JMap}
+
+import org.apache.flink.table.api._
+import org.apache.flink.table.descriptors.FormatDescriptorValidator.FORMAT_PROPERTY_VERSION
+import org.apache.flink.table.descriptors._
+import org.apache.flink.table.util.Logging
+
+import _root_.scala.collection.JavaConverters._
+import _root_.scala.collection.mutable
+
+/**
+ * Service provider interface for finding a suitable [[TableFormatFactory]] for the
+ * given properties.
+ */
+object TableFormatFactoryService extends Logging {
+
+ private lazy val defaultLoader = ServiceLoader.load(classOf[TableFormatFactory[_]])
+
+ /**
+ * Finds a table format factory of the given class and creates configured instances from the
+ * given descriptor.
+ *
+ * @param factoryClass desired format factory
+ * @param descriptor descriptor that describes the format
+ * @tparam T factory class type
+ * @return configured instance from factory
+ */
+ def find[T](factoryClass: Class[T], descriptor: Descriptor): T = {
+ find(factoryClass, descriptor, null)
+ }
+
+ /**
+ * Finds a table format factory of the given class and creates configured instances from the
+ * given descriptor and classloader.
+ *
+ * @param factoryClass desired format factory
+ * @param descriptor descriptor that describes the format
+ * @param classLoader classloader for service loading
+ * @tparam T factory class type
+ * @return configured instance from factory
+ */
+ def find[T](
+ factoryClass: Class[T],
+ descriptor: Descriptor,
+ classLoader: ClassLoader)
+ : T = {
+
+ val properties = new DescriptorProperties()
+ descriptor.addProperties(properties)
+ find(factoryClass, properties.asMap, classLoader)
+ }
+
+ /**
+ * Finds a table format factory of the given class and creates configured instances from the
+ * given property map.
+ *
+ * @param factoryClass desired format factory
+ * @param propertyMap properties that describes the format
+ * @tparam T factory class type
+ * @return configured instance from factory
+ */
+ def find[T](factoryClass: Class[T], propertyMap: JMap[String, String]): T = {
+ find(factoryClass, propertyMap, null)
+ }
+
+ /**
+ * Finds a table format factory of the given class and creates configured instances from the
+ * given property map and classloader.
+ *
+ * @param factoryClass desired format factory
+ * @param propertyMap properties that describes the format
+ * @param classLoader classloader for service loading
+ * @tparam T factory class type
+ * @return configured instance from factory
+ */
+ def find[T](
+ factoryClass: Class[T],
+ propertyMap: JMap[String, String],
+ classLoader: ClassLoader)
+ : T = {
+
+ val properties = propertyMap.asScala.toMap
+
+ // find matching context
+ val (foundFactories, contextFactories) = findMatchingContext(properties, classLoader)
+ if (contextFactories.isEmpty) {
+ throw new NoMatchingTableFormatException(
+ "No context matches.",
+ factoryClass,
+ foundFactories,
+ properties)
+ }
+
+ // filter for factory class
+ val classFactories = contextFactories.filter(f => factoryClass.isAssignableFrom(f.getClass))
+ if (classFactories.isEmpty) {
+ throw new NoMatchingTableFormatException(
+ s"No factory implements '${factoryClass.getCanonicalName}'.",
+ factoryClass,
+ foundFactories,
+ properties)
+ }
+
+ // filter by supported keys
+ val plainGivenKeys = mutable.ArrayBuffer[String]()
+ properties.keys.foreach { k =>
+ // replace arrays with wildcard
+ val key = k.replaceAll(".\\d+", ".#")
+ // ignore duplicates
+ if (!plainGivenKeys.contains(key)) {
+ plainGivenKeys += key
+ }
+ }
+ var lastKey: Option[String] = None
+ val supportedFactories = classFactories.filter { factory =>
+ val requiredContextKeys = normalizeContext(factory).keySet
+ val includeSchema = factory.supportsSchemaDerivation()
+ val supportedKeys = normalizeSupportedProperties(factory)
+ val givenKeys = plainGivenKeys
+ // ignore context keys
+ .filter(!requiredContextKeys.contains(_))
+ // ignore non-format (or schema) keys
+ .filter { k =>
+ if (includeSchema) {
+ k.startsWith(SchemaValidator.SCHEMA + ".") ||
+ k.startsWith(FormatDescriptorValidator.FORMAT + ".")
+ } else {
+ k.startsWith(FormatDescriptorValidator.FORMAT + ".")
+ }
+ }
+ givenKeys.forall { k =>
+ lastKey = Option(k)
+ supportedKeys.contains(k)
+ }
+ }
+ if (supportedFactories.isEmpty && classFactories.length == 1 && lastKey.isDefined) {
+ // special case: when there is only one matching factory but the last property key
+ // was incorrect
+ val factory = classFactories.head
+ val supportedKeys = normalizeSupportedProperties(factory)
+ throw new NoMatchingTableFormatException(
+ s"""
+ |The matching factory '${factory.getClass.getName}' doesn't support '${lastKey.get}'.
+ |
+ |Supported properties of this factory are:
+ |${supportedKeys.sorted.mkString("\n")}""".stripMargin,
+ factoryClass,
+ foundFactories,
+ properties)
+ } else if (supportedFactories.isEmpty) {
+ throw new NoMatchingTableFormatException(
+ s"No factory supports all properties.",
+ factoryClass,
+ foundFactories,
+ properties)
+ } else if (supportedFactories.length > 1) {
+ throw new AmbiguousTableFormatException(
+ supportedFactories,
+ factoryClass,
+ foundFactories,
+ properties)
+ }
+
+ supportedFactories.head.asInstanceOf[T]
+ }
+
+ private def findMatchingContext(
+ properties: Map[String, String],
+ classLoader: ClassLoader)
+ : (Seq[TableFormatFactory[_]], Seq[TableFormatFactory[_]]) = {
+
+ val foundFactories = mutable.ArrayBuffer[TableFormatFactory[_]]()
+ val matchingFactories = mutable.ArrayBuffer[TableFormatFactory[_]]()
+
+ try {
+ val iter = if (classLoader == null) {
+ defaultLoader.iterator()
+ } else {
+ val customLoader = ServiceLoader.load(classOf[TableFormatFactory[_]], classLoader)
+ customLoader.iterator()
+ }
+ while (iter.hasNext) {
+ val factory = iter.next()
+ foundFactories += factory
+
+ val requestedContext = normalizeContext(factory)
+
+ val plainContext = mutable.Map[String, String]()
+ plainContext ++= requestedContext
+ // we remove the version for now until we have the first backwards compatibility case
+ // with the version we can provide mappings in case the format changes
+ plainContext.remove(FORMAT_PROPERTY_VERSION)
+
+ // check if required context is met
+ if (plainContext.forall(e => properties.contains(e._1) && properties(e._1) == e._2)) {
+ matchingFactories += factory
+ }
+ }
+ } catch {
+ case e: ServiceConfigurationError =>
+ LOG.error("Could not load service provider for table format factories.", e)
+ throw new TableException("Could not load service provider for table format factories.", e)
+ }
+
+ (foundFactories, matchingFactories)
+ }
+
+ private def normalizeContext(factory: TableFormatFactory[_]): Map[String, String] = {
+ val requiredContextJava = factory.requiredContext()
+ if (requiredContextJava != null) {
--- End diff --
same as below
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200660774
--- Diff: flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroRowFormatFactory.java ---
@@ -0,0 +1,97 @@
+/*
+ * 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.formats.avro;
+
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.table.descriptors.AvroValidator;
+import org.apache.flink.table.descriptors.DescriptorProperties;
+import org.apache.flink.table.descriptors.FormatDescriptorValidator;
+import org.apache.flink.table.formats.DeserializationSchemaFactory;
+import org.apache.flink.table.formats.SerializationSchemaFactory;
+import org.apache.flink.types.Row;
+
+import org.apache.avro.specific.SpecificRecord;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Table format factory for providing configured instances of Avro-to-row {@link SerializationSchema}
+ * and {@link DeserializationSchema}.
+ */
+public class AvroRowFormatFactory implements SerializationSchemaFactory<Row>, DeserializationSchemaFactory<Row> {
+
+ @Override
+ public Map<String, String> requiredContext() {
+ final Map<String, String> context = new HashMap<>();
+ context.put(FormatDescriptorValidator.FORMAT_TYPE(), AvroValidator.FORMAT_TYPE_VALUE);
+ context.put(FormatDescriptorValidator.FORMAT_PROPERTY_VERSION(), "1");
+ return context;
+ }
+
+ @Override
+ public boolean supportsSchemaDerivation() {
+ return false;
+ }
+
+ @Override
+ public List<String> supportedProperties() {
+ final List<String> properties = new ArrayList<>();
+ properties.add(AvroValidator.FORMAT_RECORD_CLASS);
+ properties.add(AvroValidator.FORMAT_AVRO_SCHEMA);
+ return properties;
+ }
+
+ @Override
+ public DeserializationSchema<Row> createDeserializationSchema(Map<String, String> properties) {
+ final DescriptorProperties props = new DescriptorProperties(true);
--- End diff --
`props` -> `descriptorProperties`
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200653181
--- Diff: flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceFactoryTestBase.java ---
@@ -0,0 +1,175 @@
+/*
+ * 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.streaming.connectors.kafka;
+
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.streaming.connectors.kafka.config.StartupMode;
+import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.api.Types;
+import org.apache.flink.table.descriptors.Kafka;
+import org.apache.flink.table.descriptors.Rowtime;
+import org.apache.flink.table.descriptors.Schema;
+import org.apache.flink.table.descriptors.TestTableSourceDescriptor;
+import org.apache.flink.table.formats.utils.TestDeserializationSchema;
+import org.apache.flink.table.formats.utils.TestTableFormat;
+import org.apache.flink.table.sources.RowtimeAttributeDescriptor;
+import org.apache.flink.table.sources.TableSource;
+import org.apache.flink.table.sources.TableSourceFactoryService;
+import org.apache.flink.table.sources.tsextractors.ExistingField;
+import org.apache.flink.table.sources.wmstrategies.AscendingTimestamps;
+import org.apache.flink.types.Row;
+
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Abstract test base for {@link KafkaTableSourceFactory}.
+ */
+public abstract class KafkaTableSourceFactoryTestBase {
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testTableSource() {
+
+ // prepare parameters for Kafka table source
+
+ final TableSchema schema = TableSchema.builder()
+ .field("fruit-name", Types.STRING())
+ .field("count", Types.DECIMAL())
+ .field("event-time", Types.SQL_TIMESTAMP())
+ .field("proc-time", Types.SQL_TIMESTAMP())
+ .build();
+
+ final String proctimeAttribute = "proc-time";
+
+ final List<RowtimeAttributeDescriptor> rowtimeAttributeDescriptors = Collections.singletonList(
+ new RowtimeAttributeDescriptor("event-time", new ExistingField("time"), new AscendingTimestamps()));
+
+ final Map<String, String> fieldMapping = new HashMap<>();
+ fieldMapping.put("fruit-name", "name");
+ fieldMapping.put("count", "count");
+
+ final String topic = "myTopic";
+
+ final Properties kafkaProperties = new Properties();
+ kafkaProperties.setProperty("zookeeper.connect", "dummy");
+ kafkaProperties.setProperty("group.id", "dummy");
+
+ final Map<KafkaTopicPartition, Long> specificOffsets = new HashMap<>();
+ specificOffsets.put(new KafkaTopicPartition(topic, 0), 100L);
+ specificOffsets.put(new KafkaTopicPartition(topic, 1), 123L);
+
+ final TestDeserializationSchema deserializationSchema = new TestDeserializationSchema(
+ TableSchema.builder()
+ .field("name", Types.STRING())
+ .field("count", Types.DECIMAL())
+ .field("time", Types.SQL_TIMESTAMP())
+ .build()
+ .toRowType()
+ );
+
+ final StartupMode startupMode = StartupMode.SPECIFIC_OFFSETS;
+
+ final KafkaTableSource expected = getKafkaTableSource(
+ schema,
+ proctimeAttribute,
+ rowtimeAttributeDescriptors,
+ fieldMapping,
+ topic,
+ kafkaProperties,
+ deserializationSchema,
+ startupMode,
+ specificOffsets);
+
+ // construct table source using descriptors and table source factory
+
+ final Map<Integer, Long> offsets = new HashMap<>();
+ offsets.put(0, 100L);
+ offsets.put(1, 123L);
+
+ final TestTableSourceDescriptor testDesc = new TestTableSourceDescriptor(
+ new Kafka()
+ .version(getKafkaVersion())
+ .topic(topic)
+ .properties(kafkaProperties)
+ .startFromSpecificOffsets(offsets))
+ .addFormat(new TestTableFormat())
+ .addSchema(
+ new Schema()
+ .field("fruit-name", Types.STRING()).from("name")
+ .field("count", Types.DECIMAL()) // no from so it must match with the input
+ .field("event-time", Types.SQL_TIMESTAMP()).rowtime(
+ new Rowtime().timestampsFromField("time").watermarksPeriodicAscending())
+ .field("proc-time", Types.SQL_TIMESTAMP()).proctime());
+
+ final TableSource<?> foundSource = TableSourceFactoryService.findAndCreateTableSource(testDesc);
+
+ assertEquals(expected, foundSource);
+
+ // test Kafka consumer
+ final KafkaTableSource kafkaSource = (KafkaTableSource) foundSource;
+ final KafkaTableSource observed = spy(kafkaSource);
+ final StreamExecutionEnvironment env = mock(StreamExecutionEnvironment.class);
--- End diff --
Please use proper mocks here. Mockito tests are hard to maintain in the future.
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r201019886
--- Diff: flink-libraries/flink-table/src/test/scala/org/apache/flink/table/formats/TableFormatFactoryServiceTest.scala ---
@@ -0,0 +1,122 @@
+/*
+ * 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.formats
+
+import java.util.{HashMap => JHashMap, Map => JMap}
+
+import org.apache.flink.table.api.{AmbiguousTableFormatException, NoMatchingTableFormatException}
+import org.apache.flink.table.formats.utils.TestAmbiguousTableFormatFactory
+import org.junit.Assert.{assertNotNull, assertTrue}
+import org.junit.Test
+
+/**
+ * Tests for [[TableFormatFactoryService]].
+ */
+class TableFormatFactoryServiceTest {
+
+ @Test
+ def testValidProperties(): Unit = {
+ val props = properties()
+ assertNotNull(
+ TableFormatFactoryService.find(classOf[TableFormatFactory[_]], props))
+ }
+
+ @Test
+ def testDifferentContextVersion(): Unit = {
+ val props = properties()
+ props.put("format.property-version", "2")
+ // the format should still be found
+ assertNotNull(TableFormatFactoryService.find(classOf[TableFormatFactory[_]], props))
+ }
+
+ @Test
+ def testAmbiguousMoreSupportSelection(): Unit = {
+ val props = properties()
+ props.remove("format.important")
+ props.put("format.special_path", "/what/ever")
+ assertTrue(
+ TableFormatFactoryService
+ .find(classOf[TableFormatFactory[_]], props)
+ .isInstanceOf[TestAmbiguousTableFormatFactory])
+ }
+
+ @Test
+ def testAmbiguousClassBasedSelection(): Unit = {
+ val props = properties()
+ props.remove("format.important")
--- End diff --
rename `format.important` to something like `format.testTableFormatFactoryUniqueProperty`
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200989972
--- Diff: flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceFactoryTestBase.java ---
@@ -0,0 +1,175 @@
+/*
+ * 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.streaming.connectors.kafka;
+
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.streaming.connectors.kafka.config.StartupMode;
+import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.api.Types;
+import org.apache.flink.table.descriptors.Kafka;
+import org.apache.flink.table.descriptors.Rowtime;
+import org.apache.flink.table.descriptors.Schema;
+import org.apache.flink.table.descriptors.TestTableSourceDescriptor;
+import org.apache.flink.table.formats.utils.TestDeserializationSchema;
+import org.apache.flink.table.formats.utils.TestTableFormat;
+import org.apache.flink.table.sources.RowtimeAttributeDescriptor;
+import org.apache.flink.table.sources.TableSource;
+import org.apache.flink.table.sources.TableSourceFactoryService;
+import org.apache.flink.table.sources.tsextractors.ExistingField;
+import org.apache.flink.table.sources.wmstrategies.AscendingTimestamps;
+import org.apache.flink.types.Row;
+
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Abstract test base for {@link KafkaTableSourceFactory}.
+ */
+public abstract class KafkaTableSourceFactoryTestBase {
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testTableSource() {
+
+ // prepare parameters for Kafka table source
+
+ final TableSchema schema = TableSchema.builder()
+ .field("fruit-name", Types.STRING())
--- End diff --
deduplication of values like `"fruit-name"` within a single test + self documenting code that all of those occurrences are indeed and should be the same. Both help when debugging, refactoring and extending the test in the future.
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200662577
--- Diff: flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroRowFormatFactoryTest.java ---
@@ -0,0 +1,91 @@
+/*
+ * 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.formats.avro;
+
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.formats.avro.generated.User;
+import org.apache.flink.table.descriptors.Avro;
+import org.apache.flink.table.descriptors.Descriptor;
+import org.apache.flink.table.descriptors.DescriptorProperties;
+import org.apache.flink.table.formats.DeserializationSchemaFactory;
+import org.apache.flink.table.formats.SerializationSchemaFactory;
+import org.apache.flink.table.formats.TableFormatFactoryService;
+
+import org.junit.Test;
+
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Tests for the {@link AvroRowFormatFactory}.
+ */
+public class AvroRowFormatFactoryTest {
+
+ private static final Class<User> AVRO_SPECIFIC_RECORD = User.class;
+
+ private static final String AVRO_SCHEMA = User.getClassSchema().toString();
+
+ @Test
+ public void testRecordClass() {
+ final Map<String, String> props = toMap(new Avro().recordClass(AVRO_SPECIFIC_RECORD));
+
+ // test serialization schema
+ final SerializationSchema<?> actual1 = TableFormatFactoryService
+ .find(SerializationSchemaFactory.class, props)
+ .createSerializationSchema(props);
+ final SerializationSchema<?> expected1 = new AvroRowSerializationSchema(AVRO_SPECIFIC_RECORD);
+ assertEquals(expected1, actual1);
+
+ // test deserialization schema
+ final DeserializationSchema<?> actual2 = TableFormatFactoryService
+ .find(DeserializationSchemaFactory.class, props)
+ .createDeserializationSchema(props);
+ final AvroRowDeserializationSchema expected2 = new AvroRowDeserializationSchema(AVRO_SPECIFIC_RECORD);
+ assertEquals(expected2, actual2);
+ }
+
+ @Test
+ public void testAvroSchema() {
+ final Map<String, String> props = toMap(new Avro().avroSchema(AVRO_SCHEMA));
+
+ // test serialization schema
+ final SerializationSchema<?> actual1 = TableFormatFactoryService
+ .find(SerializationSchemaFactory.class, props)
+ .createSerializationSchema(props);
+ final SerializationSchema<?> expected1 = new AvroRowSerializationSchema(AVRO_SCHEMA);
+ assertEquals(expected1, actual1);
+
+ // test deserialization schema
+ final DeserializationSchema<?> actual2 = TableFormatFactoryService
+ .find(DeserializationSchemaFactory.class, props)
+ .createDeserializationSchema(props);
+ final AvroRowDeserializationSchema expected2 = new AvroRowDeserializationSchema(AVRO_SCHEMA);
+ assertEquals(expected2, actual2);
+ }
+
+ private Map<String, String> toMap(Descriptor... desc) {
--- End diff --
static? `props` -> `properties`
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r201309003
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/sources/DefinedFieldMapping.scala ---
@@ -50,7 +50,8 @@ trait DefinedFieldMapping {
* type. It can also provide a mapping for fields which are not in the [[TableSchema]] to make
* fields in the physical [[TypeInformation]] accessible for a [[TimestampExtractor]].
*
- * @return A mapping from [[TableSchema]] fields to [[TypeInformation]] fields.
+ * @return A mapping from [[TableSchema]] fields to [[TypeInformation]] fields or
--- End diff --
This is public API that we cannot change easily. I think the reason why Fabian chose null here is also for interoperability between Scala and Java.
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200999721
--- Diff: flink-libraries/flink-table/src/test/scala/org/apache/flink/table/descriptors/TableSourceDescriptorTest.scala ---
@@ -0,0 +1,133 @@
+/*
+ * 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.descriptors
+
+import org.apache.flink.table.api.Types
+import org.apache.flink.table.utils.TableTestBase
+import org.junit.Assert.assertEquals
+import org.junit.Test
+
+import scala.collection.JavaConverters._
+
+class TableSourceDescriptorTest extends TableTestBase {
+
+ @Test
+ def testStreamTableSourceDescriptor(): Unit = {
+ val util = streamTestUtil()
+ val desc = util.tableEnv
+ .from(
+ FileSystem()
+ .path("/path/to/csv"))
+ .withFormat(
+ Csv()
+ .field("myfield", Types.STRING)
+ .field("myfield2", Types.INT)
+ .quoteCharacter(';')
+ .fieldDelimiter("#")
+ .lineDelimiter("\r\n")
+ .commentPrefix("%%")
+ .ignoreFirstLine()
+ .ignoreParseErrors())
+ .withSchema(
+ Schema()
+ .field("myfield", Types.STRING)
+ .field("myfield2", Types.INT)
+ .field("proctime", Types.SQL_TIMESTAMP).proctime()
+ )
+ val expected = Seq(
--- End diff --
Ditto. `expectedProperties`?
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r201016561
--- Diff: flink-libraries/flink-table/src/test/scala/org/apache/flink/table/formats/TableFormatFactoryServiceTest.scala ---
@@ -0,0 +1,122 @@
+/*
+ * 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.formats
+
+import java.util.{HashMap => JHashMap, Map => JMap}
+
+import org.apache.flink.table.api.{AmbiguousTableFormatException, NoMatchingTableFormatException}
+import org.apache.flink.table.formats.utils.TestAmbiguousTableFormatFactory
--- End diff --
Am I guessing correctly, that this test is searching the class path for classes that extend `TableFormatFactory` and tries to instantiate/match them to the properties below? This reminds me too much of testing static code or code with global variables. Also it probably shows problem with encapsulation/code reusability. Shouldn't `TableFormatFactoryService` be split into following components:
1. `TableFormatFactoryDiscoverer` or even more generic `new ClassDiscoverer(classOf[TableFormatFactory])` - class that searches for the instances of `TableFormatFactory` and returns them as a list
2. `new TableFormatFactoryService(formats: Seq[Class[TableFormatFactory]])` that validates properties and instantiate format factories
3. some tiny class/code that glues 1. and 2. together
testing those components would be easier and less implicit compared to how it is now.
For example now, it is unclear to me which format factories affect this test and in what way.
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r201364539
--- Diff: flink-libraries/flink-table/src/test/scala/org/apache/flink/table/formats/utils/TestTableFormatFactory.scala ---
@@ -0,0 +1,60 @@
+/*
+ * 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.formats.utils
+
+import java.util
+
+import org.apache.flink.api.common.serialization.DeserializationSchema
+import org.apache.flink.table.descriptors.{DescriptorProperties, FormatDescriptorValidator, SchemaValidator}
+import org.apache.flink.table.formats.DeserializationSchemaFactory
+import org.apache.flink.types.Row
+
+/**
+ * Table format factory for testing.
+ */
+class TestTableFormatFactory extends DeserializationSchemaFactory[Row] {
+
+ override def requiredContext(): util.Map[String, String] = {
+ val context = new util.HashMap[String, String]()
+ context.put("format.type", "test-format")
--- End diff --
Generally, you are right. The problem of having a static variable containing a property key is however that you can break backwards compatibility and all tests automatically succeed because they all referenced the common variable.
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200997560
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/formats/TableFormatFactoryService.scala ---
@@ -0,0 +1,245 @@
+/*
+ * 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.formats
+
+import java.util.{ServiceConfigurationError, ServiceLoader, Map => JMap}
+
+import org.apache.flink.table.api._
+import org.apache.flink.table.descriptors.FormatDescriptorValidator.FORMAT_PROPERTY_VERSION
+import org.apache.flink.table.descriptors._
+import org.apache.flink.table.util.Logging
+
+import _root_.scala.collection.JavaConverters._
+import _root_.scala.collection.mutable
+
+/**
+ * Service provider interface for finding a suitable [[TableFormatFactory]] for the
+ * given properties.
+ */
+object TableFormatFactoryService extends Logging {
+
+ private lazy val defaultLoader = ServiceLoader.load(classOf[TableFormatFactory[_]])
+
+ /**
+ * Finds a table format factory of the given class and creates configured instances from the
+ * given descriptor.
+ *
+ * @param factoryClass desired format factory
+ * @param descriptor descriptor that describes the format
+ * @tparam T factory class type
+ * @return configured instance from factory
+ */
+ def find[T](factoryClass: Class[T], descriptor: Descriptor): T = {
+ find(factoryClass, descriptor, null)
+ }
+
+ /**
+ * Finds a table format factory of the given class and creates configured instances from the
+ * given descriptor and classloader.
+ *
+ * @param factoryClass desired format factory
+ * @param descriptor descriptor that describes the format
+ * @param classLoader classloader for service loading
+ * @tparam T factory class type
+ * @return configured instance from factory
+ */
+ def find[T](
+ factoryClass: Class[T],
+ descriptor: Descriptor,
+ classLoader: ClassLoader)
+ : T = {
--- End diff --
is this how we usually indent start of the function? looks a little bit weird :)
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200708088
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/descriptors/DescriptorProperties.scala ---
@@ -858,13 +859,18 @@ class DescriptorProperties(normalizeKeys: Boolean = true) {
/**
* Validates a type property.
*/
- def validateType(key: String, isOptional: Boolean): Unit = {
+ def validateType(key: String, requireRow: Boolean, isOptional: Boolean): Unit = {
if (!properties.contains(key)) {
if (!isOptional) {
throw new ValidationException(s"Could not find required property '$key'.")
}
} else {
- TypeStringUtils.readTypeInfo(properties(key)) // throws validation exceptions
+ // throws validation exceptions
--- End diff --
That we don't validate the string but let the parser do the work for us. I updated the comment.
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r201307506
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/BatchTableEnvironment.scala ---
@@ -110,16 +111,44 @@ abstract class BatchTableEnvironment(
}
}
-// TODO expose this once we have enough table source factories that can deal with it
-// /**
-// * Creates a table from a descriptor that describes the source connector, source encoding,
-// * the resulting table schema, and other properties.
-// *
-// * @param connectorDescriptor connector descriptor describing the source of the table
-// */
-// def from(connectorDescriptor: ConnectorDescriptor): BatchTableSourceDescriptor = {
-// new BatchTableSourceDescriptor(this, connectorDescriptor)
-// }
+ /**
+ * Creates a table from a descriptor that describes the source connector, the source format,
+ * the resulting table schema, and other properties.
+ *
+ * Descriptors allow for declaring communication to external systems in an
+ * implementation-agnostic way. The classpath is scanned for connectors and matching connectors
+ * are configured accordingly.
+ *
+ * The following example shows how to read from a Kafka connector using a JSON format and
+ * creating a table:
+ *
+ * {{{
+ *
+ * tableEnv
+ * .from(
+ * new Kafka()
+ * .version("0.11")
+ * .topic("clicks")
+ * .property("zookeeper.connect", "localhost")
+ * .property("group.id", "click-group")
+ * .startFromEarliest())
+ * .withFormat(
+ * new Json()
+ * .jsonSchema("{...}")
+ * .failOnMissingField(false))
+ * .withSchema(
+ * new Schema()
+ * .field("user-name", "VARCHAR").from("u_name")
+ * .field("count", "DECIMAL")
+ * .field("proc-time", "TIMESTAMP").proctime())
+ * .toTable()
+ * }}}
+ *
+ * @param connectorDescriptor connector descriptor describing the source of the table
+ */
+ def from(connectorDescriptor: ConnectorDescriptor): BatchTableSourceDescriptor = {
--- End diff --
Once we reimplement the environments in Java and introduce proper interfaces for hiding the implementation. This problem will be gone anyway.
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200982725
--- Diff: flink-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010AvroTableSource.java ---
@@ -46,7 +53,9 @@
* @param properties Properties for the Kafka consumer.
* @param schema Schema of the produced table.
* @param record Avro specific record.
+ * @deprecated Use table descriptors instead of implementation-specific builders.
*/
+ @Deprecated
--- End diff --
Yes, I assumed that it works that way as you referred and it seemed to me that deprecating the class should suffice. But you are right, it doesn't harm (except of larger commit to review)
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200665311
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/formats/TableFormatFactoryService.scala ---
@@ -0,0 +1,245 @@
+/*
+ * 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.formats
+
+import java.util.{ServiceConfigurationError, ServiceLoader, Map => JMap}
+
+import org.apache.flink.table.api._
+import org.apache.flink.table.descriptors.FormatDescriptorValidator.FORMAT_PROPERTY_VERSION
+import org.apache.flink.table.descriptors._
+import org.apache.flink.table.util.Logging
+
+import _root_.scala.collection.JavaConverters._
+import _root_.scala.collection.mutable
+
+/**
+ * Service provider interface for finding a suitable [[TableFormatFactory]] for the
+ * given properties.
+ */
+object TableFormatFactoryService extends Logging {
+
+ private lazy val defaultLoader = ServiceLoader.load(classOf[TableFormatFactory[_]])
+
+ /**
+ * Finds a table format factory of the given class and creates configured instances from the
+ * given descriptor.
+ *
+ * @param factoryClass desired format factory
+ * @param descriptor descriptor that describes the format
+ * @tparam T factory class type
+ * @return configured instance from factory
+ */
+ def find[T](factoryClass: Class[T], descriptor: Descriptor): T = {
+ find(factoryClass, descriptor, null)
+ }
+
+ /**
+ * Finds a table format factory of the given class and creates configured instances from the
+ * given descriptor and classloader.
+ *
+ * @param factoryClass desired format factory
+ * @param descriptor descriptor that describes the format
+ * @param classLoader classloader for service loading
+ * @tparam T factory class type
+ * @return configured instance from factory
+ */
+ def find[T](
+ factoryClass: Class[T],
+ descriptor: Descriptor,
+ classLoader: ClassLoader)
+ : T = {
+
+ val properties = new DescriptorProperties()
+ descriptor.addProperties(properties)
+ find(factoryClass, properties.asMap, classLoader)
+ }
+
+ /**
+ * Finds a table format factory of the given class and creates configured instances from the
+ * given property map.
+ *
+ * @param factoryClass desired format factory
+ * @param propertyMap properties that describes the format
+ * @tparam T factory class type
+ * @return configured instance from factory
+ */
+ def find[T](factoryClass: Class[T], propertyMap: JMap[String, String]): T = {
+ find(factoryClass, propertyMap, null)
+ }
+
+ /**
+ * Finds a table format factory of the given class and creates configured instances from the
+ * given property map and classloader.
+ *
+ * @param factoryClass desired format factory
+ * @param propertyMap properties that describes the format
+ * @param classLoader classloader for service loading
+ * @tparam T factory class type
+ * @return configured instance from factory
+ */
+ def find[T](
+ factoryClass: Class[T],
+ propertyMap: JMap[String, String],
+ classLoader: ClassLoader)
+ : T = {
+
+ val properties = propertyMap.asScala.toMap
+
+ // find matching context
+ val (foundFactories, contextFactories) = findMatchingContext(properties, classLoader)
+ if (contextFactories.isEmpty) {
+ throw new NoMatchingTableFormatException(
+ "No context matches.",
+ factoryClass,
+ foundFactories,
+ properties)
+ }
+
+ // filter for factory class
+ val classFactories = contextFactories.filter(f => factoryClass.isAssignableFrom(f.getClass))
+ if (classFactories.isEmpty) {
+ throw new NoMatchingTableFormatException(
+ s"No factory implements '${factoryClass.getCanonicalName}'.",
+ factoryClass,
+ foundFactories,
+ properties)
+ }
+
+ // filter by supported keys
+ val plainGivenKeys = mutable.ArrayBuffer[String]()
+ properties.keys.foreach { k =>
+ // replace arrays with wildcard
+ val key = k.replaceAll(".\\d+", ".#")
+ // ignore duplicates
+ if (!plainGivenKeys.contains(key)) {
+ plainGivenKeys += key
+ }
+ }
+ var lastKey: Option[String] = None
+ val supportedFactories = classFactories.filter { factory =>
+ val requiredContextKeys = normalizeContext(factory).keySet
+ val includeSchema = factory.supportsSchemaDerivation()
+ val supportedKeys = normalizeSupportedProperties(factory)
+ val givenKeys = plainGivenKeys
+ // ignore context keys
+ .filter(!requiredContextKeys.contains(_))
+ // ignore non-format (or schema) keys
+ .filter { k =>
+ if (includeSchema) {
+ k.startsWith(SchemaValidator.SCHEMA + ".") ||
+ k.startsWith(FormatDescriptorValidator.FORMAT + ".")
+ } else {
+ k.startsWith(FormatDescriptorValidator.FORMAT + ".")
+ }
+ }
+ givenKeys.forall { k =>
+ lastKey = Option(k)
+ supportedKeys.contains(k)
+ }
+ }
+ if (supportedFactories.isEmpty && classFactories.length == 1 && lastKey.isDefined) {
+ // special case: when there is only one matching factory but the last property key
+ // was incorrect
+ val factory = classFactories.head
+ val supportedKeys = normalizeSupportedProperties(factory)
+ throw new NoMatchingTableFormatException(
+ s"""
+ |The matching factory '${factory.getClass.getName}' doesn't support '${lastKey.get}'.
+ |
+ |Supported properties of this factory are:
+ |${supportedKeys.sorted.mkString("\n")}""".stripMargin,
+ factoryClass,
+ foundFactories,
+ properties)
+ } else if (supportedFactories.isEmpty) {
+ throw new NoMatchingTableFormatException(
+ s"No factory supports all properties.",
+ factoryClass,
+ foundFactories,
+ properties)
+ } else if (supportedFactories.length > 1) {
+ throw new AmbiguousTableFormatException(
+ supportedFactories,
+ factoryClass,
+ foundFactories,
+ properties)
+ }
+
+ supportedFactories.head.asInstanceOf[T]
+ }
+
+ private def findMatchingContext(
+ properties: Map[String, String],
+ classLoader: ClassLoader)
+ : (Seq[TableFormatFactory[_]], Seq[TableFormatFactory[_]]) = {
+
+ val foundFactories = mutable.ArrayBuffer[TableFormatFactory[_]]()
+ val matchingFactories = mutable.ArrayBuffer[TableFormatFactory[_]]()
+
+ try {
+ val iter = if (classLoader == null) {
+ defaultLoader.iterator()
+ } else {
+ val customLoader = ServiceLoader.load(classOf[TableFormatFactory[_]], classLoader)
+ customLoader.iterator()
+ }
+ while (iter.hasNext) {
+ val factory = iter.next()
+ foundFactories += factory
+
+ val requestedContext = normalizeContext(factory)
+
+ val plainContext = mutable.Map[String, String]()
+ plainContext ++= requestedContext
+ // we remove the version for now until we have the first backwards compatibility case
+ // with the version we can provide mappings in case the format changes
+ plainContext.remove(FORMAT_PROPERTY_VERSION)
+
+ // check if required context is met
+ if (plainContext.forall(e => properties.contains(e._1) && properties(e._1) == e._2)) {
+ matchingFactories += factory
+ }
+ }
+ } catch {
+ case e: ServiceConfigurationError =>
+ LOG.error("Could not load service provider for table format factories.", e)
+ throw new TableException("Could not load service provider for table format factories.", e)
+ }
+
+ (foundFactories, matchingFactories)
+ }
+
+ private def normalizeContext(factory: TableFormatFactory[_]): Map[String, String] = {
+ val requiredContextJava = factory.requiredContext()
+ if (requiredContextJava != null) {
+ requiredContextJava.asScala.map(e => (e._1.toLowerCase, e._2)).toMap
+ } else {
+ Map[String, String]()
+ }
+ }
+
+ private def normalizeSupportedProperties(factory: TableFormatFactory[_]): Seq[String] = {
+ val supportedPropertiesJava = factory.supportedProperties()
+ if (supportedPropertiesJava != null) {
--- End diff --
How can `null` happen? Either change interface to `Optional<List<...>>` or add `checkNotNull(...)` here.
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r201003919
--- Diff: flink-libraries/flink-table/src/test/scala/org/apache/flink/table/formats/TableFormatFactoryServiceTest.scala ---
@@ -0,0 +1,122 @@
+/*
+ * 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.formats
+
+import java.util.{HashMap => JHashMap, Map => JMap}
+
+import org.apache.flink.table.api.{AmbiguousTableFormatException, NoMatchingTableFormatException}
+import org.apache.flink.table.formats.utils.TestAmbiguousTableFormatFactory
+import org.junit.Assert.{assertNotNull, assertTrue}
+import org.junit.Test
+
+/**
+ * Tests for [[TableFormatFactoryService]].
+ */
+class TableFormatFactoryServiceTest {
+
+ @Test
+ def testValidProperties(): Unit = {
+ val props = properties()
+ assertNotNull(
+ TableFormatFactoryService.find(classOf[TableFormatFactory[_]], props))
+ }
+
+ @Test
+ def testDifferentContextVersion(): Unit = {
+ val props = properties()
+ props.put("format.property-version", "2")
+ // the format should still be found
+ assertNotNull(TableFormatFactoryService.find(classOf[TableFormatFactory[_]], props))
+ }
+
+ @Test
+ def testAmbiguousMoreSupportSelection(): Unit = {
+ val props = properties()
+ props.remove("format.important")
+ props.put("format.special_path", "/what/ever")
+ assertTrue(
+ TableFormatFactoryService
+ .find(classOf[TableFormatFactory[_]], props)
+ .isInstanceOf[TestAmbiguousTableFormatFactory])
+ }
+
+ @Test
+ def testAmbiguousClassBasedSelection(): Unit = {
+ val props = properties()
+ props.remove("format.important")
+ assertTrue(
+ TableFormatFactoryService
+ .find(classOf[TestAmbiguousTableFormatFactory], props)
+ .isInstanceOf[TestAmbiguousTableFormatFactory])
+ }
+
+ @Test
+ def testAmbiguousSchemaBasedSelection(): Unit = {
+ val props = properties()
+ props.remove("format.important")
+ props.put("schema.weird_field", "unknown") // this is unknown to the schema derivation factory
--- End diff --
Shouldn't we throw an error on not recognised/unused properties? Silently ignoring such properties is prone for typos.
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r201030190
--- Diff: flink-libraries/flink-table/src/test/scala/org/apache/flink/table/sources/TestWildcardFormatTableFactory.scala ---
@@ -0,0 +1,56 @@
+/*
+ * 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.sources
+
+import java.util
+
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.table.api.TableSchema
+import org.apache.flink.table.descriptors.ConnectorDescriptorValidator.{CONNECTOR_PROPERTY_VERSION, CONNECTOR_TYPE}
+import org.apache.flink.types.Row
+
+/**
+ * Table source factory for testing with a wildcard format ("format.*").
+ */
+class TestWildcardFormatTableFactory extends TableSourceFactory[Row] {
--- End diff --
Where is this class used?
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200706878
--- Diff: flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonRowFormatFactoryTest.java ---
@@ -0,0 +1,147 @@
+/*
+ * 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.formats.json;
+
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.table.api.Types;
+import org.apache.flink.table.descriptors.Descriptor;
+import org.apache.flink.table.descriptors.DescriptorProperties;
+import org.apache.flink.table.descriptors.Json;
+import org.apache.flink.table.descriptors.Schema;
+import org.apache.flink.table.formats.DeserializationSchemaFactory;
+import org.apache.flink.table.formats.SerializationSchemaFactory;
+import org.apache.flink.table.formats.TableFormatFactoryService;
+import org.apache.flink.types.Row;
+
+import org.junit.Test;
+
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Tests for the {@link JsonRowFormatFactory}.
+ */
+public class JsonRowFormatFactoryTest {
--- End diff --
Supported and unsupported properties are tested by `JsonTest`. The validator catches those cases.
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200998882
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/sources/TableSourceFactoryService.scala ---
@@ -139,9 +139,14 @@ object TableSourceFactoryService extends Logging {
Seq[String]()
}
+ // extract wildcard prefixes
--- End diff --
please extract at least both `// extract wildcard prefixes` and `// check for supported properties` into separate method (this method is already too large)
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200658490
--- Diff: flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceFactoryTestBase.java ---
@@ -0,0 +1,175 @@
+/*
+ * 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.streaming.connectors.kafka;
+
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.streaming.connectors.kafka.config.StartupMode;
+import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.api.Types;
+import org.apache.flink.table.descriptors.Kafka;
+import org.apache.flink.table.descriptors.Rowtime;
+import org.apache.flink.table.descriptors.Schema;
+import org.apache.flink.table.descriptors.TestTableSourceDescriptor;
+import org.apache.flink.table.formats.utils.TestDeserializationSchema;
+import org.apache.flink.table.formats.utils.TestTableFormat;
+import org.apache.flink.table.sources.RowtimeAttributeDescriptor;
+import org.apache.flink.table.sources.TableSource;
+import org.apache.flink.table.sources.TableSourceFactoryService;
+import org.apache.flink.table.sources.tsextractors.ExistingField;
+import org.apache.flink.table.sources.wmstrategies.AscendingTimestamps;
+import org.apache.flink.types.Row;
+
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Abstract test base for {@link KafkaTableSourceFactory}.
+ */
+public abstract class KafkaTableSourceFactoryTestBase {
--- End diff --
I'm not sure if the value of this test is big enough to pay of it's maintenance. It seems like it's testing only whether the properties are correctly passed and it's doing it so by actually coping/pasting production from the factories like `Kafka011TableSourceFactory` to the `Kafka011TableSourceFactoryTest#getKafkaTableSource`.
It more or less brings down to writing the same code twice and comparing whether the bytes are the same :(
If you really think that factories should be tested, maybe this test should be rewritten to perform assertions like:
```
assertEquals(expectedSchema, actualSource.getSchema());
```
?
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r201363632
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/formats/TableFormatFactory.scala ---
@@ -0,0 +1,85 @@
+/*
+ * 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.formats
+
+import java.util
+
+import org.apache.flink.api.common.serialization.{DeserializationSchema, SerializationSchema}
+
+/**
+ * A factory to create different table format instances. This factory is used with Java's Service
+ * Provider Interfaces (SPI) for discovering. A factory is called with a set of normalized
+ * properties that describe the desired format. The factory allows for matching to the given set of
+ * properties. See also [[SerializationSchemaFactory]] and [[DeserializationSchemaFactory]] for
+ * creating configured instances of format classes accordingly.
+ *
+ * Classes that implement this interface need to be added to the
+ * "META_INF/services/org.apache.flink.table.formats.TableFormatFactory' file of a JAR file in
+ * the current classpath to be found.
+ *
+ * @tparam T record type that the format produces or consumes
+ */
+trait TableFormatFactory[T] {
+
+ /**
+ * Specifies the context that this factory has been implemented for. The framework guarantees
+ * to only use the factory if the specified set of properties and values are met.
+ *
+ * Typical properties might be:
+ * - format.type
+ * - format.version
+ *
+ * Specified property versions allow the framework to provide backwards compatible properties
+ * in case of string format changes:
+ * - format.property-version
+ *
+ * An empty context means that the factory matches for all requests.
+ */
+ def requiredContext(): util.Map[String, String]
--- End diff --
Context defines the "context" in which the factory will be activated.
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r201360640
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/formats/TableFormatFactoryService.scala ---
@@ -0,0 +1,245 @@
+/*
+ * 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.formats
+
+import java.util.{ServiceConfigurationError, ServiceLoader, Map => JMap}
+
+import org.apache.flink.table.api._
+import org.apache.flink.table.descriptors.FormatDescriptorValidator.FORMAT_PROPERTY_VERSION
+import org.apache.flink.table.descriptors._
+import org.apache.flink.table.util.Logging
+
+import _root_.scala.collection.JavaConverters._
+import _root_.scala.collection.mutable
+
+/**
+ * Service provider interface for finding a suitable [[TableFormatFactory]] for the
+ * given properties.
+ */
+object TableFormatFactoryService extends Logging {
+
+ private lazy val defaultLoader = ServiceLoader.load(classOf[TableFormatFactory[_]])
+
+ /**
+ * Finds a table format factory of the given class and creates configured instances from the
+ * given descriptor.
+ *
+ * @param factoryClass desired format factory
--- End diff --
As mentioned in the description of this PR, I wanted to make this whole format discovery independent from particular interfaces. Maybe we will introduce block-based writer/reader interfaces or async IO interfaces. This should not require changes in the factory service anymore.
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200670463
--- Diff: flink-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010TableSource.java ---
@@ -18,48 +18,99 @@
package org.apache.flink.streaming.connectors.kafka;
-import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.streaming.connectors.kafka.config.StartupMode;
+import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.sources.RowtimeAttributeDescriptor;
import org.apache.flink.table.sources.StreamTableSource;
import org.apache.flink.types.Row;
+import java.util.List;
+import java.util.Map;
import java.util.Properties;
/**
* Kafka {@link StreamTableSource} for Kafka 0.10.
*/
-@PublicEvolving
-public abstract class Kafka010TableSource extends KafkaTableSource {
+@Internal
+public class Kafka010TableSource extends KafkaTableSource {
- // The deserialization schema for the Kafka records
- private final DeserializationSchema<Row> deserializationSchema;
+ /**
+ * Creates a Kafka 0.10 {@link StreamTableSource}.
+ *
+ * @param schema Schema of the produced table.
+ * @param proctimeAttribute Field name of the processing time attribute, null if no
--- End diff --
Replace null with `Optional`
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200668190
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/BatchTableEnvironment.scala ---
@@ -110,16 +111,44 @@ abstract class BatchTableEnvironment(
}
}
-// TODO expose this once we have enough table source factories that can deal with it
-// /**
-// * Creates a table from a descriptor that describes the source connector, source encoding,
-// * the resulting table schema, and other properties.
-// *
-// * @param connectorDescriptor connector descriptor describing the source of the table
-// */
-// def from(connectorDescriptor: ConnectorDescriptor): BatchTableSourceDescriptor = {
-// new BatchTableSourceDescriptor(this, connectorDescriptor)
-// }
+ /**
+ * Creates a table from a descriptor that describes the source connector, the source format,
+ * the resulting table schema, and other properties.
+ *
+ * Descriptors allow for declaring communication to external systems in an
+ * implementation-agnostic way. The classpath is scanned for connectors and matching connectors
+ * are configured accordingly.
+ *
+ * The following example shows how to read from a Kafka connector using a JSON format and
+ * creating a table:
+ *
+ * {{{
+ *
+ * tableEnv
+ * .from(
+ * new Kafka()
+ * .version("0.11")
+ * .topic("clicks")
+ * .property("zookeeper.connect", "localhost")
+ * .property("group.id", "click-group")
+ * .startFromEarliest())
+ * .withFormat(
+ * new Json()
+ * .jsonSchema("{...}")
+ * .failOnMissingField(false))
+ * .withSchema(
+ * new Schema()
+ * .field("user-name", "VARCHAR").from("u_name")
+ * .field("count", "DECIMAL")
+ * .field("proc-time", "TIMESTAMP").proctime())
+ * .toTable()
+ * }}}
+ *
+ * @param connectorDescriptor connector descriptor describing the source of the table
+ */
+ def from(connectorDescriptor: ConnectorDescriptor): BatchTableSourceDescriptor = {
--- End diff --
Instead of coping/pasting java doc in `StreamingTableEnvironment` and `BatchTableEnvironment`, I would prefer to use `@see` tag http://www.oracle.com/technetwork/java/javase/documentation/index-137868.html#multiple@see that would point to `TableEnvironment`
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r201592609
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/formats/TableFormatFactory.scala ---
@@ -0,0 +1,85 @@
+/*
+ * 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.formats
+
+import java.util
+
+import org.apache.flink.api.common.serialization.{DeserializationSchema, SerializationSchema}
+
+/**
+ * A factory to create different table format instances. This factory is used with Java's Service
+ * Provider Interfaces (SPI) for discovering. A factory is called with a set of normalized
+ * properties that describe the desired format. The factory allows for matching to the given set of
+ * properties. See also [[SerializationSchemaFactory]] and [[DeserializationSchemaFactory]] for
+ * creating configured instances of format classes accordingly.
+ *
+ * Classes that implement this interface need to be added to the
+ * "META_INF/services/org.apache.flink.table.formats.TableFormatFactory' file of a JAR file in
+ * the current classpath to be found.
+ *
+ * @tparam T record type that the format produces or consumes
+ */
+trait TableFormatFactory[T] {
+
+ /**
+ * Specifies the context that this factory has been implemented for. The framework guarantees
+ * to only use the factory if the specified set of properties and values are met.
+ *
+ * Typical properties might be:
+ * - format.type
+ * - format.version
+ *
+ * Specified property versions allow the framework to provide backwards compatible properties
+ * in case of string format changes:
+ * - format.property-version
+ *
+ * An empty context means that the factory matches for all requests.
+ */
+ def requiredContext(): util.Map[String, String]
--- End diff --
I still do not understand this name. Could you think about something more descriptive? It seems to me like this method returns the set of properties that are required to match given factory. Thus `requiredProperties` seems better, but maybe I'm missing something?
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200607191
--- Diff: flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceFactory.java ---
@@ -124,110 +123,130 @@
params.putProperties(properties);
// validate
- new SchemaValidator(true).validate(params);
+ // allow Kafka timestamps to be used, watermarks can not be received from source
+ new SchemaValidator(true, supportsKafkaTimestamps(), false).validate(params);
new KafkaValidator().validate(params);
- formatValidator().validate(params);
- // build
- final KafkaTableSource.Builder builder = createBuilderWithFormat(params);
+ // deserialization schema using format discovery
+ final DeserializationSchemaFactory<?> formatFactory = TableFormatFactoryService.find(
+ DeserializationSchemaFactory.class,
+ properties,
+ this.getClass().getClassLoader());
+ @SuppressWarnings("unchecked")
+ final DeserializationSchema<Row> deserializationSchema = (DeserializationSchema<Row>) formatFactory
+ .createDeserializationSchema(properties);
- // topic
- final String topic = params.getString(CONNECTOR_TOPIC);
- builder.forTopic(topic);
+ // schema
--- End diff --
Here are again lots of unnecessary changes clogging review :(
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200999836
--- Diff: flink-libraries/flink-table/src/test/scala/org/apache/flink/table/descriptors/TableSourceDescriptorTest.scala ---
@@ -0,0 +1,133 @@
+/*
+ * 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.descriptors
+
+import org.apache.flink.table.api.Types
+import org.apache.flink.table.utils.TableTestBase
+import org.junit.Assert.assertEquals
+import org.junit.Test
+
+import scala.collection.JavaConverters._
+
+class TableSourceDescriptorTest extends TableTestBase {
+
+ @Test
+ def testStreamTableSourceDescriptor(): Unit = {
+ val util = streamTestUtil()
+ val desc = util.tableEnv
+ .from(
+ FileSystem()
+ .path("/path/to/csv"))
+ .withFormat(
+ Csv()
+ .field("myfield", Types.STRING)
+ .field("myfield2", Types.INT)
+ .quoteCharacter(';')
+ .fieldDelimiter("#")
+ .lineDelimiter("\r\n")
+ .commentPrefix("%%")
+ .ignoreFirstLine()
+ .ignoreParseErrors())
+ .withSchema(
+ Schema()
+ .field("myfield", Types.STRING)
+ .field("myfield2", Types.INT)
+ .field("proctime", Types.SQL_TIMESTAMP).proctime()
+ )
+ val expected = Seq(
+ "connector.property-version" -> "1",
+ "connector.type" -> "filesystem",
+ "connector.path" -> "/path/to/csv",
+ "format.property-version" -> "1",
+ "format.type" -> "csv",
+ "format.fields.0.name" -> "myfield",
+ "format.fields.0.type" -> "VARCHAR",
+ "format.fields.1.name" -> "myfield2",
+ "format.fields.1.type" -> "INT",
+ "format.quote-character" -> ";",
+ "format.field-delimiter" -> "#",
+ "format.line-delimiter" -> "\r\n",
+ "format.comment-prefix" -> "%%",
+ "format.ignore-first-line" -> "true",
+ "format.ignore-parse-errors" -> "true",
+ "schema.0.name" -> "myfield",
+ "schema.0.type" -> "VARCHAR",
+ "schema.1.name" -> "myfield2",
+ "schema.1.type" -> "INT",
+ "schema.2.name" -> "proctime",
+ "schema.2.type" -> "TIMESTAMP",
+ "schema.2.proctime" -> "true"
+ ).toMap
+
+ val actual = new DescriptorProperties(true)
+ desc.addProperties(actual)
+
+ assertEquals(expected.asJava, actual.asMap)
+ }
+
+ @Test
+ def testBatchTableSourceDescriptor(): Unit = {
+ val util = batchTestUtil()
+ val desc = util.tableEnv
+ .from(
+ FileSystem()
+ .path("/path/to/csv"))
+ .withFormat(
+ Csv()
+ .field("myfield", Types.STRING)
+ .field("myfield2", Types.INT)
+ .quoteCharacter(';')
+ .fieldDelimiter("#")
+ .lineDelimiter("\r\n")
+ .commentPrefix("%%")
+ .ignoreFirstLine()
+ .ignoreParseErrors())
+ .withSchema(
+ Schema()
+ .field("myfield", Types.STRING)
+ .field("myfield2", Types.INT)
+ )
+ val expected = Seq(
--- End diff --
ditto
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200651952
--- Diff: flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceFactoryTestBase.java ---
@@ -0,0 +1,175 @@
+/*
+ * 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.streaming.connectors.kafka;
+
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.streaming.connectors.kafka.config.StartupMode;
+import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.api.Types;
+import org.apache.flink.table.descriptors.Kafka;
+import org.apache.flink.table.descriptors.Rowtime;
+import org.apache.flink.table.descriptors.Schema;
+import org.apache.flink.table.descriptors.TestTableSourceDescriptor;
+import org.apache.flink.table.formats.utils.TestDeserializationSchema;
+import org.apache.flink.table.formats.utils.TestTableFormat;
+import org.apache.flink.table.sources.RowtimeAttributeDescriptor;
+import org.apache.flink.table.sources.TableSource;
+import org.apache.flink.table.sources.TableSourceFactoryService;
+import org.apache.flink.table.sources.tsextractors.ExistingField;
+import org.apache.flink.table.sources.wmstrategies.AscendingTimestamps;
+import org.apache.flink.types.Row;
+
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Abstract test base for {@link KafkaTableSourceFactory}.
+ */
+public abstract class KafkaTableSourceFactoryTestBase {
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testTableSource() {
+
+ // prepare parameters for Kafka table source
+
+ final TableSchema schema = TableSchema.builder()
+ .field("fruit-name", Types.STRING())
+ .field("count", Types.DECIMAL())
+ .field("event-time", Types.SQL_TIMESTAMP())
+ .field("proc-time", Types.SQL_TIMESTAMP())
+ .build();
+
+ final String proctimeAttribute = "proc-time";
+
+ final List<RowtimeAttributeDescriptor> rowtimeAttributeDescriptors = Collections.singletonList(
+ new RowtimeAttributeDescriptor("event-time", new ExistingField("time"), new AscendingTimestamps()));
+
+ final Map<String, String> fieldMapping = new HashMap<>();
+ fieldMapping.put("fruit-name", "name");
+ fieldMapping.put("count", "count");
+
+ final String topic = "myTopic";
+
+ final Properties kafkaProperties = new Properties();
+ kafkaProperties.setProperty("zookeeper.connect", "dummy");
+ kafkaProperties.setProperty("group.id", "dummy");
+
+ final Map<KafkaTopicPartition, Long> specificOffsets = new HashMap<>();
+ specificOffsets.put(new KafkaTopicPartition(topic, 0), 100L);
+ specificOffsets.put(new KafkaTopicPartition(topic, 1), 123L);
+
+ final TestDeserializationSchema deserializationSchema = new TestDeserializationSchema(
+ TableSchema.builder()
+ .field("name", Types.STRING())
+ .field("count", Types.DECIMAL())
+ .field("time", Types.SQL_TIMESTAMP())
+ .build()
+ .toRowType()
+ );
+
+ final StartupMode startupMode = StartupMode.SPECIFIC_OFFSETS;
+
+ final KafkaTableSource expected = getKafkaTableSource(
+ schema,
+ proctimeAttribute,
+ rowtimeAttributeDescriptors,
+ fieldMapping,
+ topic,
+ kafkaProperties,
+ deserializationSchema,
+ startupMode,
+ specificOffsets);
+
+ // construct table source using descriptors and table source factory
+
+ final Map<Integer, Long> offsets = new HashMap<>();
+ offsets.put(0, 100L);
+ offsets.put(1, 123L);
+
+ final TestTableSourceDescriptor testDesc = new TestTableSourceDescriptor(
+ new Kafka()
+ .version(getKafkaVersion())
+ .topic(topic)
+ .properties(kafkaProperties)
+ .startFromSpecificOffsets(offsets))
+ .addFormat(new TestTableFormat())
+ .addSchema(
+ new Schema()
+ .field("fruit-name", Types.STRING()).from("name")
+ .field("count", Types.DECIMAL()) // no from so it must match with the input
+ .field("event-time", Types.SQL_TIMESTAMP()).rowtime(
+ new Rowtime().timestampsFromField("time").watermarksPeriodicAscending())
+ .field("proc-time", Types.SQL_TIMESTAMP()).proctime());
+
+ final TableSource<?> foundSource = TableSourceFactoryService.findAndCreateTableSource(testDesc);
+
+ assertEquals(expected, foundSource);
+
+ // test Kafka consumer
+ final KafkaTableSource kafkaSource = (KafkaTableSource) foundSource;
+ final KafkaTableSource observed = spy(kafkaSource);
+ final StreamExecutionEnvironment env = mock(StreamExecutionEnvironment.class);
+ when(env.addSource(any(SourceFunction.class))).thenReturn(mock(DataStreamSource.class));
+ observed.getDataStream(env);
+
+ verify(env).addSource(any(getFlinkKafkaConsumer()));
+
+ verify(observed).getKafkaConsumer(
+ eq(topic),
+ // the consumer modifies the properties that's why we compare them this way
+ eq(kafkaSource.getProperties()),
+ eq(deserializationSchema));
+ }
+
+ // --------------------------------------------------------------------------------------------
+ // For version-specific tests
+ // --------------------------------------------------------------------------------------------
+
+ protected abstract String getKafkaVersion();
+
+ protected abstract Class<FlinkKafkaConsumerBase<Row>> getFlinkKafkaConsumer();
+
+ protected abstract KafkaTableSource getKafkaTableSource(
--- End diff --
`getExpectedKafkaTableSource`
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200598387
--- Diff: flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSource.java ---
@@ -55,50 +57,101 @@
*/
@Internal
public abstract class KafkaTableSource
- implements StreamTableSource<Row>, DefinedProctimeAttribute, DefinedRowtimeAttributes {
+ implements StreamTableSource<Row>, DefinedProctimeAttribute, DefinedRowtimeAttributes, DefinedFieldMapping {
--- End diff --
nit: split interfaces into new lines?
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200608975
--- Diff: flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceFactory.java ---
@@ -124,110 +123,130 @@
params.putProperties(properties);
// validate
- new SchemaValidator(true).validate(params);
+ // allow Kafka timestamps to be used, watermarks can not be received from source
+ new SchemaValidator(true, supportsKafkaTimestamps(), false).validate(params);
new KafkaValidator().validate(params);
- formatValidator().validate(params);
- // build
- final KafkaTableSource.Builder builder = createBuilderWithFormat(params);
+ // deserialization schema using format discovery
+ final DeserializationSchemaFactory<?> formatFactory = TableFormatFactoryService.find(
+ DeserializationSchemaFactory.class,
+ properties,
+ this.getClass().getClassLoader());
+ @SuppressWarnings("unchecked")
+ final DeserializationSchema<Row> deserializationSchema = (DeserializationSchema<Row>) formatFactory
+ .createDeserializationSchema(properties);
- // topic
- final String topic = params.getString(CONNECTOR_TOPIC);
- builder.forTopic(topic);
+ // schema
+ final TableSchema schema = params.getTableSchema(SCHEMA());
+
+ // proctime
+ final String proctimeAttribute = SchemaValidator.deriveProctimeAttribute(params).orElse(null);
+
+ // rowtime
+ final List<RowtimeAttributeDescriptor> rowtimeAttributes = SchemaValidator.deriveRowtimeAttributes(params);
+ if (rowtimeAttributes.size() > 1) {
--- End diff --
Do we need to keep this check here? This is checked later in `TableSourceUtil` and code between here and there supports `rowtimeAttributes.size() > 1`, so this is duplicated logic.
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r201017476
--- Diff: flink-libraries/flink-table/src/test/scala/org/apache/flink/table/formats/TableFormatFactoryServiceTest.scala ---
@@ -0,0 +1,122 @@
+/*
+ * 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.formats
+
+import java.util.{HashMap => JHashMap, Map => JMap}
+
+import org.apache.flink.table.api.{AmbiguousTableFormatException, NoMatchingTableFormatException}
+import org.apache.flink.table.formats.utils.TestAmbiguousTableFormatFactory
+import org.junit.Assert.{assertNotNull, assertTrue}
+import org.junit.Test
+
+/**
+ * Tests for [[TableFormatFactoryService]].
+ */
+class TableFormatFactoryServiceTest {
+
+ @Test
+ def testValidProperties(): Unit = {
+ val props = properties()
+ assertNotNull(
+ TableFormatFactoryService.find(classOf[TableFormatFactory[_]], props))
+ }
+
+ @Test
+ def testDifferentContextVersion(): Unit = {
+ val props = properties()
+ props.put("format.property-version", "2")
--- End diff --
? factories should provide forward compatibility?
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r201026925
--- Diff: flink-libraries/flink-table/src/test/scala/org/apache/flink/table/formats/TableFormatFactoryServiceTest.scala ---
@@ -0,0 +1,122 @@
+/*
+ * 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.formats
+
+import java.util.{HashMap => JHashMap, Map => JMap}
+
+import org.apache.flink.table.api.{AmbiguousTableFormatException, NoMatchingTableFormatException}
+import org.apache.flink.table.formats.utils.TestAmbiguousTableFormatFactory
+import org.junit.Assert.{assertNotNull, assertTrue}
+import org.junit.Test
+
+/**
+ * Tests for [[TableFormatFactoryService]].
+ */
+class TableFormatFactoryServiceTest {
+
+ @Test
+ def testValidProperties(): Unit = {
+ val props = properties()
+ assertNotNull(
+ TableFormatFactoryService.find(classOf[TableFormatFactory[_]], props))
+ }
+
+ @Test
+ def testDifferentContextVersion(): Unit = {
+ val props = properties()
+ props.put("format.property-version", "2")
+ // the format should still be found
+ assertNotNull(TableFormatFactoryService.find(classOf[TableFormatFactory[_]], props))
+ }
+
+ @Test
+ def testAmbiguousMoreSupportSelection(): Unit = {
+ val props = properties()
+ props.remove("format.important")
+ props.put("format.special_path", "/what/ever")
+ assertTrue(
+ TableFormatFactoryService
+ .find(classOf[TableFormatFactory[_]], props)
+ .isInstanceOf[TestAmbiguousTableFormatFactory])
+ }
+
+ @Test
+ def testAmbiguousClassBasedSelection(): Unit = {
+ val props = properties()
+ props.remove("format.important")
+ assertTrue(
+ TableFormatFactoryService
+ .find(classOf[TestAmbiguousTableFormatFactory], props)
+ .isInstanceOf[TestAmbiguousTableFormatFactory])
+ }
+
+ @Test
+ def testAmbiguousSchemaBasedSelection(): Unit = {
+ val props = properties()
+ props.remove("format.important")
+ props.put("schema.weird_field", "unknown") // this is unknown to the schema derivation factory
+ assertTrue(
+ TableFormatFactoryService
+ .find(classOf[TableFormatFactory[_]], props)
+ .isInstanceOf[TestAmbiguousTableFormatFactory])
+ }
+
+ @Test(expected = classOf[NoMatchingTableFormatException])
+ def testMissingClass(): Unit = {
+ val props = properties()
+ // this class is not a valid factory
+ TableFormatFactoryService.find(classOf[TableFormatFactoryServiceTest], props)
+ }
+
+ @Test(expected = classOf[NoMatchingTableFormatException])
+ def testInvalidContext(): Unit = {
+ val props = properties()
+ props.put("format.type", "FAIL") // no context specifies this
--- End diff --
`s/FAIL/unknown_format_type/g` and remove comment?
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r201020483
--- Diff: flink-libraries/flink-table/src/test/scala/org/apache/flink/table/formats/utils/TestAmbiguousTableFormatFactory.scala ---
@@ -16,22 +16,32 @@
* limitations under the License.
*/
-package org.apache.flink.streaming.connectors.kafka;
+package org.apache.flink.table.formats.utils
-import static org.apache.flink.table.descriptors.KafkaValidator.CONNECTOR_VERSION_VALUE_010;
+import java.util
+
+import org.apache.flink.table.formats.TableFormatFactory
+import org.apache.flink.types.Row
/**
- * Tests for {@link Kafka010AvroTableSourceFactory}.
- */
-public class Kafka010AvroTableSourceFactoryTest extends KafkaAvroTableSourceFactoryTestBase {
+ * Table format factory for testing.
+ */
+class TestAmbiguousTableFormatFactory extends TableFormatFactory[Row] {
+
+ override def requiredContext(): util.Map[String, String] = {
+ val context = new util.HashMap[String, String]()
+ context.put("format.type", "test-format")
+ context.put("format.property-version", "1")
+ context
+ }
- @Override
- protected String version() {
- return CONNECTOR_VERSION_VALUE_010;
- }
+ override def supportsSchemaDerivation(): Boolean = false // no schema derivation
- @Override
- protected KafkaAvroTableSource.Builder builder() {
- return Kafka010AvroTableSource.builder();
- }
+ override def supportedProperties(): util.List[String] = {
+ val properties = new util.ArrayList[String]()
+ // has no 'format.important'
--- End diff --
move this comment to javadoc of the class
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200678922
--- Diff: flink-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010JsonTableSource.java ---
@@ -32,7 +33,13 @@
/**
* Kafka {@link StreamTableSource} for Kafka 0.10.
+ *
+ * @deprecated Use the {@link org.apache.flink.table.descriptors.Kafka} descriptor together
+ * with descriptors for schema and format instead. Descriptors allow for
+ * implementation-agnostic definition of tables. See also
+ * {@link org.apache.flink.table.api.TableEnvironment#from(ConnectorDescriptor)}.
*/
+@Deprecated
--- End diff --
`PublicEvolving` is not `Public` so we can modify them if necessary. Additionally, this class should never have gotten this annotation as we do not guarantee API stability for the Table API at the moment.
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r201344640
--- Diff: flink-libraries/flink-table/src/test/scala/org/apache/flink/table/formats/TableFormatFactoryServiceTest.scala ---
@@ -0,0 +1,122 @@
+/*
+ * 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.formats
+
+import java.util.{HashMap => JHashMap, Map => JMap}
+
+import org.apache.flink.table.api.{AmbiguousTableFormatException, NoMatchingTableFormatException}
+import org.apache.flink.table.formats.utils.TestAmbiguousTableFormatFactory
+import org.junit.Assert.{assertNotNull, assertTrue}
+import org.junit.Test
+
+/**
+ * Tests for [[TableFormatFactoryService]].
+ */
+class TableFormatFactoryServiceTest {
+
+ @Test
+ def testValidProperties(): Unit = {
+ val props = properties()
+ assertNotNull(
+ TableFormatFactoryService.find(classOf[TableFormatFactory[_]], props))
+ }
+
+ @Test
+ def testDifferentContextVersion(): Unit = {
+ val props = properties()
+ props.put("format.property-version", "2")
+ // the format should still be found
+ assertNotNull(TableFormatFactoryService.find(classOf[TableFormatFactory[_]], props))
+ }
+
+ @Test
+ def testAmbiguousMoreSupportSelection(): Unit = {
+ val props = properties()
+ props.remove("format.important")
+ props.put("format.special_path", "/what/ever")
+ assertTrue(
+ TableFormatFactoryService
+ .find(classOf[TableFormatFactory[_]], props)
+ .isInstanceOf[TestAmbiguousTableFormatFactory])
+ }
+
+ @Test
+ def testAmbiguousClassBasedSelection(): Unit = {
+ val props = properties()
+ props.remove("format.important")
+ assertTrue(
+ TableFormatFactoryService
+ .find(classOf[TestAmbiguousTableFormatFactory], props)
+ .isInstanceOf[TestAmbiguousTableFormatFactory])
+ }
+
+ @Test
+ def testAmbiguousSchemaBasedSelection(): Unit = {
+ val props = properties()
+ props.remove("format.important")
+ props.put("schema.weird_field", "unknown") // this is unknown to the schema derivation factory
--- End diff --
The format with schema derivation feels not responsible because of this field. But since there is another format that feels responsible no exception is thrown. If `TestAmbiguousTableFormatFactory` would not exist an exception would have been thrown.
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200993332
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/BatchTableEnvironment.scala ---
@@ -110,16 +111,44 @@ abstract class BatchTableEnvironment(
}
}
-// TODO expose this once we have enough table source factories that can deal with it
-// /**
-// * Creates a table from a descriptor that describes the source connector, source encoding,
-// * the resulting table schema, and other properties.
-// *
-// * @param connectorDescriptor connector descriptor describing the source of the table
-// */
-// def from(connectorDescriptor: ConnectorDescriptor): BatchTableSourceDescriptor = {
-// new BatchTableSourceDescriptor(this, connectorDescriptor)
-// }
+ /**
+ * Creates a table from a descriptor that describes the source connector, the source format,
+ * the resulting table schema, and other properties.
+ *
+ * Descriptors allow for declaring communication to external systems in an
+ * implementation-agnostic way. The classpath is scanned for connectors and matching connectors
+ * are configured accordingly.
+ *
+ * The following example shows how to read from a Kafka connector using a JSON format and
+ * creating a table:
+ *
+ * {{{
+ *
+ * tableEnv
+ * .from(
+ * new Kafka()
+ * .version("0.11")
+ * .topic("clicks")
+ * .property("zookeeper.connect", "localhost")
+ * .property("group.id", "click-group")
+ * .startFromEarliest())
+ * .withFormat(
+ * new Json()
+ * .jsonSchema("{...}")
+ * .failOnMissingField(false))
+ * .withSchema(
+ * new Schema()
+ * .field("user-name", "VARCHAR").from("u_name")
+ * .field("count", "DECIMAL")
+ * .field("proc-time", "TIMESTAMP").proctime())
+ * .toTable()
+ * }}}
+ *
+ * @param connectorDescriptor connector descriptor describing the source of the table
+ */
+ def from(connectorDescriptor: ConnectorDescriptor): BatchTableSourceDescriptor = {
--- End diff --
I'm not sure. Maintaining duplicated comment in three places, especially one that's lie this one that non trivial, is a futile task. They will drift out of sync. For me the only question is If we can avoid it in some acceptable way.
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200679446
--- Diff: flink-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010AvroTableSource.java ---
@@ -46,7 +53,9 @@
* @param properties Properties for the Kafka consumer.
* @param schema Schema of the produced table.
* @param record Avro specific record.
+ * @deprecated Use table descriptors instead of implementation-specific builders.
*/
+ @Deprecated
--- End diff --
https://stackoverflow.com/questions/17615019/class-with-deprecated-annotation-means-that-all-methods-and-fields-automaticall
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200998167
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/sources/DefinedFieldMapping.scala ---
@@ -50,7 +50,8 @@ trait DefinedFieldMapping {
* type. It can also provide a mapping for fields which are not in the [[TableSchema]] to make
* fields in the physical [[TypeInformation]] accessible for a [[TimestampExtractor]].
*
- * @return A mapping from [[TableSchema]] fields to [[TypeInformation]] fields.
+ * @return A mapping from [[TableSchema]] fields to [[TypeInformation]] fields or
--- End diff --
`Optional` :(
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200662272
--- Diff: flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroRowFormatFactoryTest.java ---
@@ -0,0 +1,91 @@
+/*
+ * 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.formats.avro;
+
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.formats.avro.generated.User;
+import org.apache.flink.table.descriptors.Avro;
+import org.apache.flink.table.descriptors.Descriptor;
+import org.apache.flink.table.descriptors.DescriptorProperties;
+import org.apache.flink.table.formats.DeserializationSchemaFactory;
+import org.apache.flink.table.formats.SerializationSchemaFactory;
+import org.apache.flink.table.formats.TableFormatFactoryService;
+
+import org.junit.Test;
+
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Tests for the {@link AvroRowFormatFactory}.
+ */
+public class AvroRowFormatFactoryTest {
+
+ private static final Class<User> AVRO_SPECIFIC_RECORD = User.class;
+
+ private static final String AVRO_SCHEMA = User.getClassSchema().toString();
+
+ @Test
+ public void testRecordClass() {
+ final Map<String, String> props = toMap(new Avro().recordClass(AVRO_SPECIFIC_RECORD));
+
+ // test serialization schema
+ final SerializationSchema<?> actual1 = TableFormatFactoryService
+ .find(SerializationSchemaFactory.class, props)
+ .createSerializationSchema(props);
+ final SerializationSchema<?> expected1 = new AvroRowSerializationSchema(AVRO_SPECIFIC_RECORD);
+ assertEquals(expected1, actual1);
+
+ // test deserialization schema
+ final DeserializationSchema<?> actual2 = TableFormatFactoryService
+ .find(DeserializationSchemaFactory.class, props)
+ .createDeserializationSchema(props);
+ final AvroRowDeserializationSchema expected2 = new AvroRowDeserializationSchema(AVRO_SPECIFIC_RECORD);
+ assertEquals(expected2, actual2);
+ }
+
+ @Test
+ public void testAvroSchema() {
+ final Map<String, String> props = toMap(new Avro().avroSchema(AVRO_SCHEMA));
+
+ // test serialization schema
+ final SerializationSchema<?> actual1 = TableFormatFactoryService
+ .find(SerializationSchemaFactory.class, props)
+ .createSerializationSchema(props);
+ final SerializationSchema<?> expected1 = new AvroRowSerializationSchema(AVRO_SCHEMA);
+ assertEquals(expected1, actual1);
+
+ // test deserialization schema
--- End diff --
ditto
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r201359834
--- Diff: flink-libraries/flink-table/src/test/scala/org/apache/flink/table/formats/utils/TestAmbiguousTableFormatFactory.scala ---
@@ -16,22 +16,32 @@
* limitations under the License.
*/
-package org.apache.flink.streaming.connectors.kafka;
+package org.apache.flink.table.formats.utils
-import static org.apache.flink.table.descriptors.KafkaValidator.CONNECTOR_VERSION_VALUE_010;
+import java.util
+
+import org.apache.flink.table.formats.TableFormatFactory
+import org.apache.flink.types.Row
/**
- * Tests for {@link Kafka010AvroTableSourceFactory}.
- */
-public class Kafka010AvroTableSourceFactoryTest extends KafkaAvroTableSourceFactoryTestBase {
+ * Table format factory for testing.
+ */
+class TestAmbiguousTableFormatFactory extends TableFormatFactory[Row] {
--- End diff --
This implicit interconnection is on purposes. A similar configuration could happen in a SQL Client library folder. The `TableFormatFactoryServiceTest` should fail if they are changed too much. But I added more comments for implementers. If `TestAmbiguousTableFormatFactory` would extend from `TestTableFormatFactory` then the class based selection would not work anymore.
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r201350063
--- Diff: flink-libraries/flink-table/src/test/scala/org/apache/flink/table/formats/TableFormatFactoryServiceTest.scala ---
@@ -0,0 +1,122 @@
+/*
+ * 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.formats
+
+import java.util.{HashMap => JHashMap, Map => JMap}
+
+import org.apache.flink.table.api.{AmbiguousTableFormatException, NoMatchingTableFormatException}
+import org.apache.flink.table.formats.utils.TestAmbiguousTableFormatFactory
--- End diff --
Yes, we could split the logic here. Would it be ok for you do do this in a follow-up issue? I guess we can combine the logic with `TableSourceFactory` and the upcoming sink discovery. We could introduce a general `FactoryDiscoverer` for all interfaces.
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200690291
--- Diff: flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceFactoryTestBase.java ---
@@ -0,0 +1,175 @@
+/*
+ * 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.streaming.connectors.kafka;
+
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.streaming.connectors.kafka.config.StartupMode;
+import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.api.Types;
+import org.apache.flink.table.descriptors.Kafka;
+import org.apache.flink.table.descriptors.Rowtime;
+import org.apache.flink.table.descriptors.Schema;
+import org.apache.flink.table.descriptors.TestTableSourceDescriptor;
+import org.apache.flink.table.formats.utils.TestDeserializationSchema;
+import org.apache.flink.table.formats.utils.TestTableFormat;
+import org.apache.flink.table.sources.RowtimeAttributeDescriptor;
+import org.apache.flink.table.sources.TableSource;
+import org.apache.flink.table.sources.TableSourceFactoryService;
+import org.apache.flink.table.sources.tsextractors.ExistingField;
+import org.apache.flink.table.sources.wmstrategies.AscendingTimestamps;
+import org.apache.flink.types.Row;
+
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Abstract test base for {@link KafkaTableSourceFactory}.
+ */
+public abstract class KafkaTableSourceFactoryTestBase {
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testTableSource() {
+
+ // prepare parameters for Kafka table source
+
+ final TableSchema schema = TableSchema.builder()
+ .field("fruit-name", Types.STRING())
--- End diff --
I don't know if moving every field here into a static variable does improve readability or maintainability. What is the benefit of doing it in this test case?
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200652221
--- Diff: flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceFactoryTestBase.java ---
@@ -0,0 +1,175 @@
+/*
+ * 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.streaming.connectors.kafka;
+
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.streaming.connectors.kafka.config.StartupMode;
+import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.api.Types;
+import org.apache.flink.table.descriptors.Kafka;
+import org.apache.flink.table.descriptors.Rowtime;
+import org.apache.flink.table.descriptors.Schema;
+import org.apache.flink.table.descriptors.TestTableSourceDescriptor;
+import org.apache.flink.table.formats.utils.TestDeserializationSchema;
+import org.apache.flink.table.formats.utils.TestTableFormat;
+import org.apache.flink.table.sources.RowtimeAttributeDescriptor;
+import org.apache.flink.table.sources.TableSource;
+import org.apache.flink.table.sources.TableSourceFactoryService;
+import org.apache.flink.table.sources.tsextractors.ExistingField;
+import org.apache.flink.table.sources.wmstrategies.AscendingTimestamps;
+import org.apache.flink.types.Row;
+
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Abstract test base for {@link KafkaTableSourceFactory}.
+ */
+public abstract class KafkaTableSourceFactoryTestBase {
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testTableSource() {
+
+ // prepare parameters for Kafka table source
+
+ final TableSchema schema = TableSchema.builder()
+ .field("fruit-name", Types.STRING())
--- End diff --
please extract the constants to `private static final` variables
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200653352
--- Diff: flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceFactoryTestBase.java ---
@@ -0,0 +1,175 @@
+/*
+ * 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.streaming.connectors.kafka;
+
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.streaming.connectors.kafka.config.StartupMode;
+import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.api.Types;
+import org.apache.flink.table.descriptors.Kafka;
+import org.apache.flink.table.descriptors.Rowtime;
+import org.apache.flink.table.descriptors.Schema;
+import org.apache.flink.table.descriptors.TestTableSourceDescriptor;
+import org.apache.flink.table.formats.utils.TestDeserializationSchema;
+import org.apache.flink.table.formats.utils.TestTableFormat;
+import org.apache.flink.table.sources.RowtimeAttributeDescriptor;
+import org.apache.flink.table.sources.TableSource;
+import org.apache.flink.table.sources.TableSourceFactoryService;
+import org.apache.flink.table.sources.tsextractors.ExistingField;
+import org.apache.flink.table.sources.wmstrategies.AscendingTimestamps;
+import org.apache.flink.types.Row;
+
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Abstract test base for {@link KafkaTableSourceFactory}.
+ */
+public abstract class KafkaTableSourceFactoryTestBase {
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testTableSource() {
+
+ // prepare parameters for Kafka table source
+
+ final TableSchema schema = TableSchema.builder()
+ .field("fruit-name", Types.STRING())
+ .field("count", Types.DECIMAL())
+ .field("event-time", Types.SQL_TIMESTAMP())
+ .field("proc-time", Types.SQL_TIMESTAMP())
+ .build();
+
+ final String proctimeAttribute = "proc-time";
+
+ final List<RowtimeAttributeDescriptor> rowtimeAttributeDescriptors = Collections.singletonList(
+ new RowtimeAttributeDescriptor("event-time", new ExistingField("time"), new AscendingTimestamps()));
+
+ final Map<String, String> fieldMapping = new HashMap<>();
+ fieldMapping.put("fruit-name", "name");
+ fieldMapping.put("count", "count");
+
+ final String topic = "myTopic";
+
+ final Properties kafkaProperties = new Properties();
+ kafkaProperties.setProperty("zookeeper.connect", "dummy");
+ kafkaProperties.setProperty("group.id", "dummy");
+
+ final Map<KafkaTopicPartition, Long> specificOffsets = new HashMap<>();
+ specificOffsets.put(new KafkaTopicPartition(topic, 0), 100L);
+ specificOffsets.put(new KafkaTopicPartition(topic, 1), 123L);
+
+ final TestDeserializationSchema deserializationSchema = new TestDeserializationSchema(
+ TableSchema.builder()
+ .field("name", Types.STRING())
+ .field("count", Types.DECIMAL())
+ .field("time", Types.SQL_TIMESTAMP())
+ .build()
+ .toRowType()
+ );
+
+ final StartupMode startupMode = StartupMode.SPECIFIC_OFFSETS;
+
+ final KafkaTableSource expected = getKafkaTableSource(
+ schema,
+ proctimeAttribute,
+ rowtimeAttributeDescriptors,
+ fieldMapping,
+ topic,
+ kafkaProperties,
+ deserializationSchema,
+ startupMode,
+ specificOffsets);
+
+ // construct table source using descriptors and table source factory
+
+ final Map<Integer, Long> offsets = new HashMap<>();
+ offsets.put(0, 100L);
+ offsets.put(1, 123L);
+
+ final TestTableSourceDescriptor testDesc = new TestTableSourceDescriptor(
+ new Kafka()
+ .version(getKafkaVersion())
+ .topic(topic)
+ .properties(kafkaProperties)
+ .startFromSpecificOffsets(offsets))
+ .addFormat(new TestTableFormat())
+ .addSchema(
+ new Schema()
+ .field("fruit-name", Types.STRING()).from("name")
+ .field("count", Types.DECIMAL()) // no from so it must match with the input
+ .field("event-time", Types.SQL_TIMESTAMP()).rowtime(
+ new Rowtime().timestampsFromField("time").watermarksPeriodicAscending())
+ .field("proc-time", Types.SQL_TIMESTAMP()).proctime());
+
+ final TableSource<?> foundSource = TableSourceFactoryService.findAndCreateTableSource(testDesc);
--- End diff --
`%s/foundSource/actualSource/g`
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200661652
--- Diff: flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroRowFormatFactoryTest.java ---
@@ -0,0 +1,91 @@
+/*
+ * 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.formats.avro;
+
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.formats.avro.generated.User;
+import org.apache.flink.table.descriptors.Avro;
+import org.apache.flink.table.descriptors.Descriptor;
+import org.apache.flink.table.descriptors.DescriptorProperties;
+import org.apache.flink.table.formats.DeserializationSchemaFactory;
+import org.apache.flink.table.formats.SerializationSchemaFactory;
+import org.apache.flink.table.formats.TableFormatFactoryService;
+
+import org.junit.Test;
+
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Tests for the {@link AvroRowFormatFactory}.
+ */
+public class AvroRowFormatFactoryTest {
+
+ private static final Class<User> AVRO_SPECIFIC_RECORD = User.class;
+
+ private static final String AVRO_SCHEMA = User.getClassSchema().toString();
+
+ @Test
+ public void testRecordClass() {
+ final Map<String, String> props = toMap(new Avro().recordClass(AVRO_SPECIFIC_RECORD));
--- End diff --
`props` -> `properties` and extract to `private static final`
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r201308428
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/formats/TableFormatFactoryService.scala ---
@@ -0,0 +1,245 @@
+/*
+ * 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.formats
+
+import java.util.{ServiceConfigurationError, ServiceLoader, Map => JMap}
+
+import org.apache.flink.table.api._
+import org.apache.flink.table.descriptors.FormatDescriptorValidator.FORMAT_PROPERTY_VERSION
+import org.apache.flink.table.descriptors._
+import org.apache.flink.table.util.Logging
+
+import _root_.scala.collection.JavaConverters._
+import _root_.scala.collection.mutable
+
+/**
+ * Service provider interface for finding a suitable [[TableFormatFactory]] for the
+ * given properties.
+ */
+object TableFormatFactoryService extends Logging {
+
+ private lazy val defaultLoader = ServiceLoader.load(classOf[TableFormatFactory[_]])
+
+ /**
+ * Finds a table format factory of the given class and creates configured instances from the
+ * given descriptor.
+ *
+ * @param factoryClass desired format factory
+ * @param descriptor descriptor that describes the format
+ * @tparam T factory class type
+ * @return configured instance from factory
+ */
+ def find[T](factoryClass: Class[T], descriptor: Descriptor): T = {
+ find(factoryClass, descriptor, null)
+ }
+
+ /**
+ * Finds a table format factory of the given class and creates configured instances from the
+ * given descriptor and classloader.
+ *
+ * @param factoryClass desired format factory
+ * @param descriptor descriptor that describes the format
+ * @param classLoader classloader for service loading
+ * @tparam T factory class type
+ * @return configured instance from factory
+ */
+ def find[T](
+ factoryClass: Class[T],
+ descriptor: Descriptor,
+ classLoader: ClassLoader)
+ : T = {
--- End diff --
I follow this styling guide: `https://github.com/databricks/scala-style-guide#indent`.
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200992037
--- Diff: flink-formats/flink-json/src/main/java/org/apache/flink/formats/json/JsonRowFormatFactory.java ---
@@ -0,0 +1,106 @@
+/*
+ * 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.formats.json;
+
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.table.descriptors.DescriptorProperties;
+import org.apache.flink.table.descriptors.FormatDescriptorValidator;
+import org.apache.flink.table.descriptors.JsonValidator;
+import org.apache.flink.table.descriptors.SchemaValidator;
+import org.apache.flink.table.formats.DeserializationSchemaFactory;
+import org.apache.flink.table.formats.SerializationSchemaFactory;
+import org.apache.flink.types.Row;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Table format factory for providing configured instances of JSON-to-row {@link SerializationSchema}
+ * and {@link DeserializationSchema}.
+ */
+public class JsonRowFormatFactory implements SerializationSchemaFactory<Row>, DeserializationSchemaFactory<Row> {
+
+ @Override
+ public Map<String, String> requiredContext() {
+ final Map<String, String> context = new HashMap<>();
+ context.put(FormatDescriptorValidator.FORMAT_TYPE(), JsonValidator.FORMAT_TYPE_VALUE);
+ context.put(FormatDescriptorValidator.FORMAT_PROPERTY_VERSION(), "1");
+ return context;
+ }
+
+ @Override
+ public boolean supportsSchemaDerivation() {
+ return true;
+ }
+
+ @Override
+ public List<String> supportedProperties() {
+ final List<String> properties = new ArrayList<>();
+ properties.add(JsonValidator.FORMAT_JSON_SCHEMA);
+ properties.add(JsonValidator.FORMAT_SCHEMA);
+ properties.add(JsonValidator.FORMAT_FAIL_ON_MISSING_FIELD);
+ properties.add(FormatDescriptorValidator.FORMAT_DERIVE_SCHEMA());
+ properties.addAll(SchemaValidator.getSchemaDerivationKeys());
+ return properties;
+ }
+
+ @Override
+ public DeserializationSchema<Row> createDeserializationSchema(Map<String, String> properties) {
+ final DescriptorProperties props = new DescriptorProperties(true);
--- End diff --
ditto regarding `props` and extracting duplicated code
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200702383
--- Diff: flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroRowFormatFactoryTest.java ---
@@ -0,0 +1,91 @@
+/*
+ * 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.formats.avro;
+
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.formats.avro.generated.User;
+import org.apache.flink.table.descriptors.Avro;
+import org.apache.flink.table.descriptors.Descriptor;
+import org.apache.flink.table.descriptors.DescriptorProperties;
+import org.apache.flink.table.formats.DeserializationSchemaFactory;
+import org.apache.flink.table.formats.SerializationSchemaFactory;
+import org.apache.flink.table.formats.TableFormatFactoryService;
+
+import org.junit.Test;
+
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Tests for the {@link AvroRowFormatFactory}.
+ */
+public class AvroRowFormatFactoryTest {
--- End diff --
I added it to the factory tests. I don't know what the general policy is. I think we don't need it for every test.
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r201029821
--- Diff: flink-libraries/flink-table/src/test/scala/org/apache/flink/table/formats/utils/TestTableFormatFactory.scala ---
@@ -0,0 +1,60 @@
+/*
+ * 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.formats.utils
+
+import java.util
+
+import org.apache.flink.api.common.serialization.DeserializationSchema
+import org.apache.flink.table.descriptors.{DescriptorProperties, FormatDescriptorValidator, SchemaValidator}
+import org.apache.flink.table.formats.DeserializationSchemaFactory
+import org.apache.flink.types.Row
+
+/**
+ * Table format factory for testing.
+ */
+class TestTableFormatFactory extends DeserializationSchemaFactory[Row] {
+
+ override def requiredContext(): util.Map[String, String] = {
+ val context = new util.HashMap[String, String]()
+ context.put("format.type", "test-format")
--- End diff --
I think you should extract or reuse already extracted constants for all of those strings like `format.type` `format.path` `format.important` etc. Generally speaking if a constant occurs more then once in code, it should be extracted.
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200670793
--- Diff: flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSource.java ---
@@ -55,50 +57,101 @@
*/
@Internal
public abstract class KafkaTableSource
- implements StreamTableSource<Row>, DefinedProctimeAttribute, DefinedRowtimeAttributes {
+ implements StreamTableSource<Row>, DefinedProctimeAttribute, DefinedRowtimeAttributes, DefinedFieldMapping {
+
+ // common table source attributes
+ // TODO make all attributes final once we drop support for format-specific table sources
/** The schema of the table. */
private final TableSchema schema;
+ /** Field name of the processing time attribute, null if no processing time field is defined. */
+ private String proctimeAttribute;
+
+ /** Descriptor for a rowtime attribute. */
+ private List<RowtimeAttributeDescriptor> rowtimeAttributeDescriptors;
+
+ /** Mapping for the fields of the table schema to fields of the physical returned type or null. */
+ private Map<String, String> fieldMapping;
--- End diff --
`Optional<Map<...>> fieldMapping`. Same goes for other nullable fields/parameters.
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r201021369
--- Diff: flink-libraries/flink-table/src/test/scala/org/apache/flink/table/formats/utils/TestAmbiguousTableFormatFactory.scala ---
@@ -16,22 +16,32 @@
* limitations under the License.
*/
-package org.apache.flink.streaming.connectors.kafka;
+package org.apache.flink.table.formats.utils
-import static org.apache.flink.table.descriptors.KafkaValidator.CONNECTOR_VERSION_VALUE_010;
+import java.util
+
+import org.apache.flink.table.formats.TableFormatFactory
+import org.apache.flink.types.Row
/**
- * Tests for {@link Kafka010AvroTableSourceFactory}.
- */
-public class Kafka010AvroTableSourceFactoryTest extends KafkaAvroTableSourceFactoryTestBase {
+ * Table format factory for testing.
+ */
+class TestAmbiguousTableFormatFactory extends TableFormatFactory[Row] {
--- End diff --
shouldn't this extend or reuse `TestTableFormatFactory`? They are now interconnected implicitly - you assume some dependencies between those two classes - and this can veerryy easily brake (silently) if someone modifies only one of those classes forgetting about the other. (silently - tests might still pass while they will not be testing what you intended anymore).
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200664426
--- Diff: flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonRowFormatFactoryTest.java ---
@@ -0,0 +1,147 @@
+/*
+ * 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.formats.json;
+
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.table.api.Types;
+import org.apache.flink.table.descriptors.Descriptor;
+import org.apache.flink.table.descriptors.DescriptorProperties;
+import org.apache.flink.table.descriptors.Json;
+import org.apache.flink.table.descriptors.Schema;
+import org.apache.flink.table.formats.DeserializationSchemaFactory;
+import org.apache.flink.table.formats.SerializationSchemaFactory;
+import org.apache.flink.table.formats.TableFormatFactoryService;
+import org.apache.flink.types.Row;
+
+import org.junit.Test;
+
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Tests for the {@link JsonRowFormatFactory}.
+ */
+public class JsonRowFormatFactoryTest {
+
+ private static final String JSON_SCHEMA =
+ "{" +
+ " 'title': 'Fruit'," +
+ " 'type': 'object'," +
+ " 'properties': {" +
+ " 'name': {" +
+ " 'type': 'string'" +
+ " }," +
+ " 'count': {" +
+ " 'type': 'integer'" +
+ " }," +
+ " 'time': {" +
+ " 'description': 'row time'," +
+ " 'type': 'string'," +
+ " 'format': 'date-time'" +
+ " }" +
+ " }," +
+ " 'required': ['name', 'count', 'time']" +
+ "}";
+
+ private static final TypeInformation<Row> SCHEMA = Types.ROW(
+ new String[]{"field1", "field2"},
+ new TypeInformation[]{Types.BOOLEAN(), Types.INT()});
+
+ @Test
+ public void testSchema() {
+ final Map<String, String> props = toMap(
--- End diff --
same comments like in previous test
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200669741
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/formats/TableFormatFactoryService.scala ---
@@ -0,0 +1,245 @@
+/*
+ * 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.formats
+
+import java.util.{ServiceConfigurationError, ServiceLoader, Map => JMap}
+
+import org.apache.flink.table.api._
+import org.apache.flink.table.descriptors.FormatDescriptorValidator.FORMAT_PROPERTY_VERSION
+import org.apache.flink.table.descriptors._
+import org.apache.flink.table.util.Logging
+
+import _root_.scala.collection.JavaConverters._
+import _root_.scala.collection.mutable
+
+/**
+ * Service provider interface for finding a suitable [[TableFormatFactory]] for the
+ * given properties.
+ */
+object TableFormatFactoryService extends Logging {
+
+ private lazy val defaultLoader = ServiceLoader.load(classOf[TableFormatFactory[_]])
+
+ /**
+ * Finds a table format factory of the given class and creates configured instances from the
+ * given descriptor.
+ *
+ * @param factoryClass desired format factory
+ * @param descriptor descriptor that describes the format
+ * @tparam T factory class type
+ * @return configured instance from factory
+ */
+ def find[T](factoryClass: Class[T], descriptor: Descriptor): T = {
+ find(factoryClass, descriptor, null)
+ }
+
+ /**
+ * Finds a table format factory of the given class and creates configured instances from the
+ * given descriptor and classloader.
+ *
+ * @param factoryClass desired format factory
+ * @param descriptor descriptor that describes the format
+ * @param classLoader classloader for service loading
+ * @tparam T factory class type
+ * @return configured instance from factory
+ */
+ def find[T](
+ factoryClass: Class[T],
+ descriptor: Descriptor,
+ classLoader: ClassLoader)
+ : T = {
+
+ val properties = new DescriptorProperties()
+ descriptor.addProperties(properties)
+ find(factoryClass, properties.asMap, classLoader)
+ }
+
+ /**
+ * Finds a table format factory of the given class and creates configured instances from the
+ * given property map.
+ *
+ * @param factoryClass desired format factory
+ * @param propertyMap properties that describes the format
+ * @tparam T factory class type
+ * @return configured instance from factory
+ */
+ def find[T](factoryClass: Class[T], propertyMap: JMap[String, String]): T = {
+ find(factoryClass, propertyMap, null)
+ }
+
+ /**
+ * Finds a table format factory of the given class and creates configured instances from the
+ * given property map and classloader.
+ *
+ * @param factoryClass desired format factory
+ * @param propertyMap properties that describes the format
+ * @param classLoader classloader for service loading
+ * @tparam T factory class type
+ * @return configured instance from factory
+ */
+ def find[T](
--- End diff --
split this method into smaller ones. Basically everywhere where you typed a comment
```
//foo bar baz
some;
code;
block;
```
replace the comment with method call:
```
def fooBarBaz() {
some;
code;
block;
}
```
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200662445
--- Diff: flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroRowFormatFactoryTest.java ---
@@ -0,0 +1,91 @@
+/*
+ * 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.formats.avro;
+
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.formats.avro.generated.User;
+import org.apache.flink.table.descriptors.Avro;
+import org.apache.flink.table.descriptors.Descriptor;
+import org.apache.flink.table.descriptors.DescriptorProperties;
+import org.apache.flink.table.formats.DeserializationSchemaFactory;
+import org.apache.flink.table.formats.SerializationSchemaFactory;
+import org.apache.flink.table.formats.TableFormatFactoryService;
+
+import org.junit.Test;
+
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Tests for the {@link AvroRowFormatFactory}.
+ */
+public class AvroRowFormatFactoryTest {
--- End diff --
`extends TestLogger` (ditto other places)
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r201593060
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/formats/TableFormatFactoryService.scala ---
@@ -0,0 +1,245 @@
+/*
+ * 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.formats
+
+import java.util.{ServiceConfigurationError, ServiceLoader, Map => JMap}
+
+import org.apache.flink.table.api._
+import org.apache.flink.table.descriptors.FormatDescriptorValidator.FORMAT_PROPERTY_VERSION
+import org.apache.flink.table.descriptors._
+import org.apache.flink.table.util.Logging
+
+import _root_.scala.collection.JavaConverters._
+import _root_.scala.collection.mutable
+
+/**
+ * Service provider interface for finding a suitable [[TableFormatFactory]] for the
+ * given properties.
+ */
+object TableFormatFactoryService extends Logging {
+
+ private lazy val defaultLoader = ServiceLoader.load(classOf[TableFormatFactory[_]])
+
+ /**
+ * Finds a table format factory of the given class and creates configured instances from the
+ * given descriptor.
+ *
+ * @param factoryClass desired format factory
+ * @param descriptor descriptor that describes the format
+ * @tparam T factory class type
+ * @return configured instance from factory
+ */
+ def find[T](factoryClass: Class[T], descriptor: Descriptor): T = {
+ find(factoryClass, descriptor, null)
+ }
+
+ /**
+ * Finds a table format factory of the given class and creates configured instances from the
+ * given descriptor and classloader.
+ *
+ * @param factoryClass desired format factory
+ * @param descriptor descriptor that describes the format
+ * @param classLoader classloader for service loading
+ * @tparam T factory class type
+ * @return configured instance from factory
+ */
+ def find[T](
+ factoryClass: Class[T],
+ descriptor: Descriptor,
+ classLoader: ClassLoader)
+ : T = {
+
+ val properties = new DescriptorProperties()
+ descriptor.addProperties(properties)
+ find(factoryClass, properties.asMap, classLoader)
+ }
+
+ /**
+ * Finds a table format factory of the given class and creates configured instances from the
+ * given property map.
+ *
+ * @param factoryClass desired format factory
+ * @param propertyMap properties that describes the format
+ * @tparam T factory class type
+ * @return configured instance from factory
+ */
+ def find[T](factoryClass: Class[T], propertyMap: JMap[String, String]): T = {
+ find(factoryClass, propertyMap, null)
+ }
+
+ /**
+ * Finds a table format factory of the given class and creates configured instances from the
+ * given property map and classloader.
+ *
+ * @param factoryClass desired format factory
+ * @param propertyMap properties that describes the format
+ * @param classLoader classloader for service loading
+ * @tparam T factory class type
+ * @return configured instance from factory
+ */
+ def find[T](
+ factoryClass: Class[T],
+ propertyMap: JMap[String, String],
+ classLoader: ClassLoader)
+ : T = {
+
+ val properties = propertyMap.asScala.toMap
+
+ // find matching context
+ val (foundFactories, contextFactories) = findMatchingContext(properties, classLoader)
+ if (contextFactories.isEmpty) {
+ throw new NoMatchingTableFormatException(
+ "No context matches.",
+ factoryClass,
+ foundFactories,
+ properties)
+ }
+
+ // filter for factory class
+ val classFactories = contextFactories.filter(f => factoryClass.isAssignableFrom(f.getClass))
+ if (classFactories.isEmpty) {
+ throw new NoMatchingTableFormatException(
+ s"No factory implements '${factoryClass.getCanonicalName}'.",
+ factoryClass,
+ foundFactories,
+ properties)
+ }
+
+ // filter by supported keys
+ val plainGivenKeys = mutable.ArrayBuffer[String]()
+ properties.keys.foreach { k =>
+ // replace arrays with wildcard
+ val key = k.replaceAll(".\\d+", ".#")
+ // ignore duplicates
+ if (!plainGivenKeys.contains(key)) {
+ plainGivenKeys += key
+ }
+ }
+ var lastKey: Option[String] = None
+ val supportedFactories = classFactories.filter { factory =>
+ val requiredContextKeys = normalizeContext(factory).keySet
+ val includeSchema = factory.supportsSchemaDerivation()
+ val supportedKeys = normalizeSupportedProperties(factory)
+ val givenKeys = plainGivenKeys
+ // ignore context keys
+ .filter(!requiredContextKeys.contains(_))
+ // ignore non-format (or schema) keys
+ .filter { k =>
+ if (includeSchema) {
+ k.startsWith(SchemaValidator.SCHEMA + ".") ||
+ k.startsWith(FormatDescriptorValidator.FORMAT + ".")
+ } else {
+ k.startsWith(FormatDescriptorValidator.FORMAT + ".")
+ }
+ }
+ givenKeys.forall { k =>
+ lastKey = Option(k)
+ supportedKeys.contains(k)
+ }
+ }
+ if (supportedFactories.isEmpty && classFactories.length == 1 && lastKey.isDefined) {
+ // special case: when there is only one matching factory but the last property key
+ // was incorrect
+ val factory = classFactories.head
+ val supportedKeys = normalizeSupportedProperties(factory)
+ throw new NoMatchingTableFormatException(
+ s"""
+ |The matching factory '${factory.getClass.getName}' doesn't support '${lastKey.get}'.
+ |
+ |Supported properties of this factory are:
+ |${supportedKeys.sorted.mkString("\n")}""".stripMargin,
+ factoryClass,
+ foundFactories,
+ properties)
+ } else if (supportedFactories.isEmpty) {
+ throw new NoMatchingTableFormatException(
+ s"No factory supports all properties.",
+ factoryClass,
+ foundFactories,
+ properties)
+ } else if (supportedFactories.length > 1) {
+ throw new AmbiguousTableFormatException(
+ supportedFactories,
+ factoryClass,
+ foundFactories,
+ properties)
+ }
+
+ supportedFactories.head.asInstanceOf[T]
+ }
+
+ private def findMatchingContext(
+ properties: Map[String, String],
+ classLoader: ClassLoader)
+ : (Seq[TableFormatFactory[_]], Seq[TableFormatFactory[_]]) = {
+
+ val foundFactories = mutable.ArrayBuffer[TableFormatFactory[_]]()
+ val matchingFactories = mutable.ArrayBuffer[TableFormatFactory[_]]()
+
+ try {
+ val iter = if (classLoader == null) {
+ defaultLoader.iterator()
+ } else {
+ val customLoader = ServiceLoader.load(classOf[TableFormatFactory[_]], classLoader)
+ customLoader.iterator()
+ }
+ while (iter.hasNext) {
+ val factory = iter.next()
+ foundFactories += factory
+
+ val requestedContext = normalizeContext(factory)
+
+ val plainContext = mutable.Map[String, String]()
+ plainContext ++= requestedContext
+ // we remove the version for now until we have the first backwards compatibility case
+ // with the version we can provide mappings in case the format changes
+ plainContext.remove(FORMAT_PROPERTY_VERSION)
+
+ // check if required context is met
+ if (plainContext.forall(e => properties.contains(e._1) && properties(e._1) == e._2)) {
+ matchingFactories += factory
+ }
+ }
+ } catch {
+ case e: ServiceConfigurationError =>
+ LOG.error("Could not load service provider for table format factories.", e)
+ throw new TableException("Could not load service provider for table format factories.", e)
+ }
+
+ (foundFactories, matchingFactories)
+ }
+
+ private def normalizeContext(factory: TableFormatFactory[_]): Map[String, String] = {
+ val requiredContextJava = factory.requiredContext()
+ if (requiredContextJava != null) {
+ requiredContextJava.asScala.map(e => (e._1.toLowerCase, e._2)).toMap
+ } else {
+ Map[String, String]()
--- End diff --
`checkNotNull(requiredContextJava)`? The interface doesn't seem to allow for nulls.
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r201002716
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/formats/TableFormatFactoryService.scala ---
@@ -0,0 +1,245 @@
+/*
+ * 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.formats
+
+import java.util.{ServiceConfigurationError, ServiceLoader, Map => JMap}
+
+import org.apache.flink.table.api._
+import org.apache.flink.table.descriptors.FormatDescriptorValidator.FORMAT_PROPERTY_VERSION
+import org.apache.flink.table.descriptors._
+import org.apache.flink.table.util.Logging
+
+import _root_.scala.collection.JavaConverters._
+import _root_.scala.collection.mutable
+
+/**
+ * Service provider interface for finding a suitable [[TableFormatFactory]] for the
+ * given properties.
+ */
+object TableFormatFactoryService extends Logging {
+
+ private lazy val defaultLoader = ServiceLoader.load(classOf[TableFormatFactory[_]])
+
+ /**
+ * Finds a table format factory of the given class and creates configured instances from the
+ * given descriptor.
+ *
+ * @param factoryClass desired format factory
+ * @param descriptor descriptor that describes the format
+ * @tparam T factory class type
+ * @return configured instance from factory
+ */
+ def find[T](factoryClass: Class[T], descriptor: Descriptor): T = {
+ find(factoryClass, descriptor, null)
--- End diff --
`Option[ClassLoader]` - same argument as before regarding nulls
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r201025967
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/formats/TableFormatFactoryService.scala ---
@@ -0,0 +1,245 @@
+/*
+ * 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.formats
+
+import java.util.{ServiceConfigurationError, ServiceLoader, Map => JMap}
+
+import org.apache.flink.table.api._
+import org.apache.flink.table.descriptors.FormatDescriptorValidator.FORMAT_PROPERTY_VERSION
+import org.apache.flink.table.descriptors._
+import org.apache.flink.table.util.Logging
+
+import _root_.scala.collection.JavaConverters._
+import _root_.scala.collection.mutable
+
+/**
+ * Service provider interface for finding a suitable [[TableFormatFactory]] for the
+ * given properties.
+ */
+object TableFormatFactoryService extends Logging {
+
+ private lazy val defaultLoader = ServiceLoader.load(classOf[TableFormatFactory[_]])
+
+ /**
+ * Finds a table format factory of the given class and creates configured instances from the
+ * given descriptor.
+ *
+ * @param factoryClass desired format factory
--- End diff --
why do we need this param? Shouldn't this be handled by either to instances of this service for serialisation and deserialisation or by copying the providing two sets of methods like `~findSerialisationFactory` and `~findDeserialisationFactory`?
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200662208
--- Diff: flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroRowFormatFactoryTest.java ---
@@ -0,0 +1,91 @@
+/*
+ * 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.formats.avro;
+
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.formats.avro.generated.User;
+import org.apache.flink.table.descriptors.Avro;
+import org.apache.flink.table.descriptors.Descriptor;
+import org.apache.flink.table.descriptors.DescriptorProperties;
+import org.apache.flink.table.formats.DeserializationSchemaFactory;
+import org.apache.flink.table.formats.SerializationSchemaFactory;
+import org.apache.flink.table.formats.TableFormatFactoryService;
+
+import org.junit.Test;
+
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Tests for the {@link AvroRowFormatFactory}.
+ */
+public class AvroRowFormatFactoryTest {
+
+ private static final Class<User> AVRO_SPECIFIC_RECORD = User.class;
+
+ private static final String AVRO_SCHEMA = User.getClassSchema().toString();
+
+ @Test
+ public void testRecordClass() {
+ final Map<String, String> props = toMap(new Avro().recordClass(AVRO_SPECIFIC_RECORD));
+
+ // test serialization schema
+ final SerializationSchema<?> actual1 = TableFormatFactoryService
+ .find(SerializationSchemaFactory.class, props)
+ .createSerializationSchema(props);
+ final SerializationSchema<?> expected1 = new AvroRowSerializationSchema(AVRO_SPECIFIC_RECORD);
+ assertEquals(expected1, actual1);
+
+ // test deserialization schema
--- End diff --
split the test into two as you have split it by comments:
`testRecordClassDeserializationSchema` and `testRecordClassSerializationSchema`
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r201338537
--- Diff: flink-libraries/flink-table/src/test/scala/org/apache/flink/table/formats/TableFormatFactoryServiceTest.scala ---
@@ -0,0 +1,122 @@
+/*
+ * 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.formats
+
+import java.util.{HashMap => JHashMap, Map => JMap}
+
+import org.apache.flink.table.api.{AmbiguousTableFormatException, NoMatchingTableFormatException}
+import org.apache.flink.table.formats.utils.TestAmbiguousTableFormatFactory
+import org.junit.Assert.{assertNotNull, assertTrue}
+import org.junit.Test
+
+/**
+ * Tests for [[TableFormatFactoryService]].
+ */
+class TableFormatFactoryServiceTest {
+
+ @Test
+ def testValidProperties(): Unit = {
+ val props = properties()
+ assertNotNull(
--- End diff --
No it should never return null.
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r201017796
--- Diff: flink-libraries/flink-table/src/test/scala/org/apache/flink/table/formats/TableFormatFactoryServiceTest.scala ---
@@ -0,0 +1,122 @@
+/*
+ * 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.formats
+
+import java.util.{HashMap => JHashMap, Map => JMap}
+
+import org.apache.flink.table.api.{AmbiguousTableFormatException, NoMatchingTableFormatException}
+import org.apache.flink.table.formats.utils.TestAmbiguousTableFormatFactory
+import org.junit.Assert.{assertNotNull, assertTrue}
+import org.junit.Test
+
+/**
+ * Tests for [[TableFormatFactoryService]].
+ */
+class TableFormatFactoryServiceTest {
+
+ @Test
+ def testValidProperties(): Unit = {
+ val props = properties()
+ assertNotNull(
+ TableFormatFactoryService.find(classOf[TableFormatFactory[_]], props))
+ }
+
+ @Test
+ def testDifferentContextVersion(): Unit = {
+ val props = properties()
+ props.put("format.property-version", "2")
+ // the format should still be found
+ assertNotNull(TableFormatFactoryService.find(classOf[TableFormatFactory[_]], props))
--- End diff --
shouldn't you test what was found?
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200610493
--- Diff: flink-connectors/flink-connector-kafka-0.11/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka011JsonTableSourceFactory.java ---
@@ -1,37 +0,0 @@
-/*
- * 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.streaming.connectors.kafka;
-
-import static org.apache.flink.table.descriptors.KafkaValidator.CONNECTOR_VERSION_VALUE_011;
-
-/**
- * Factory for creating configured instances of {@link Kafka011JsonTableSource}.
- */
-public class Kafka011JsonTableSourceFactory extends KafkaJsonTableSourceFactory {
--- End diff --
What was the purpose of this class? Was it part of a public api and was used by a user?
I kind of don't understand why this class was removed while `Kafka011JsonTableSource` was marked as `@Deprecated`.
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r201027539
--- Diff: flink-libraries/flink-table/src/test/scala/org/apache/flink/table/formats/TableFormatFactoryServiceTest.scala ---
@@ -0,0 +1,122 @@
+/*
+ * 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.formats
+
+import java.util.{HashMap => JHashMap, Map => JMap}
+
+import org.apache.flink.table.api.{AmbiguousTableFormatException, NoMatchingTableFormatException}
+import org.apache.flink.table.formats.utils.TestAmbiguousTableFormatFactory
+import org.junit.Assert.{assertNotNull, assertTrue}
+import org.junit.Test
+
+/**
+ * Tests for [[TableFormatFactoryService]].
+ */
+class TableFormatFactoryServiceTest {
+
+ @Test
+ def testValidProperties(): Unit = {
+ val props = properties()
+ assertNotNull(
+ TableFormatFactoryService.find(classOf[TableFormatFactory[_]], props))
+ }
+
+ @Test
+ def testDifferentContextVersion(): Unit = {
+ val props = properties()
+ props.put("format.property-version", "2")
+ // the format should still be found
+ assertNotNull(TableFormatFactoryService.find(classOf[TableFormatFactory[_]], props))
+ }
+
+ @Test
+ def testAmbiguousMoreSupportSelection(): Unit = {
+ val props = properties()
+ props.remove("format.important")
+ props.put("format.special_path", "/what/ever")
+ assertTrue(
+ TableFormatFactoryService
+ .find(classOf[TableFormatFactory[_]], props)
+ .isInstanceOf[TestAmbiguousTableFormatFactory])
+ }
+
+ @Test
+ def testAmbiguousClassBasedSelection(): Unit = {
+ val props = properties()
+ props.remove("format.important")
+ assertTrue(
+ TableFormatFactoryService
+ .find(classOf[TestAmbiguousTableFormatFactory], props)
+ .isInstanceOf[TestAmbiguousTableFormatFactory])
+ }
+
+ @Test
+ def testAmbiguousSchemaBasedSelection(): Unit = {
+ val props = properties()
+ props.remove("format.important")
+ props.put("schema.weird_field", "unknown") // this is unknown to the schema derivation factory
+ assertTrue(
+ TableFormatFactoryService
+ .find(classOf[TableFormatFactory[_]], props)
+ .isInstanceOf[TestAmbiguousTableFormatFactory])
+ }
+
+ @Test(expected = classOf[NoMatchingTableFormatException])
+ def testMissingClass(): Unit = {
+ val props = properties()
+ // this class is not a valid factory
+ TableFormatFactoryService.find(classOf[TableFormatFactoryServiceTest], props)
+ }
+
+ @Test(expected = classOf[NoMatchingTableFormatException])
+ def testInvalidContext(): Unit = {
+ val props = properties()
+ props.put("format.type", "FAIL") // no context specifies this
+ TableFormatFactoryService.find(classOf[TableFormatFactory[_]], props)
+ }
+
+ @Test(expected = classOf[NoMatchingTableFormatException])
+ def testUnsupportedProperty(): Unit = {
+ val props = properties()
+ props.put("format.path_new", "/new/path") // no factory has this
+ TableFormatFactoryService.find(classOf[TableFormatFactory[_]], props)
+ }
+
+ @Test(expected = classOf[AmbiguousTableFormatException])
+ def testAmbiguousFactory(): Unit = {
--- End diff --
how does this differ from `testAmbiguousSchemaBasedSelection`?
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200983274
--- Diff: flink-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010JsonTableSource.java ---
@@ -32,7 +33,13 @@
/**
* Kafka {@link StreamTableSource} for Kafka 0.10.
+ *
+ * @deprecated Use the {@link org.apache.flink.table.descriptors.Kafka} descriptor together
+ * with descriptors for schema and format instead. Descriptors allow for
+ * implementation-agnostic definition of tables. See also
+ * {@link org.apache.flink.table.api.TableEnvironment#from(ConnectorDescriptor)}.
*/
+@Deprecated
--- End diff --
As we talked offline. I wasn't sure if we could drop `PublicEvolving` classes instead of deprecating them. But if you want to deprecate them for one release, it might be better idea to do so.
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200603964
--- Diff: flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceFactory.java ---
@@ -124,110 +123,130 @@
params.putProperties(properties);
// validate
- new SchemaValidator(true).validate(params);
+ // allow Kafka timestamps to be used, watermarks can not be received from source
+ new SchemaValidator(true, supportsKafkaTimestamps(), false).validate(params);
new KafkaValidator().validate(params);
- formatValidator().validate(params);
- // build
- final KafkaTableSource.Builder builder = createBuilderWithFormat(params);
--- End diff --
Why was the `Builder` dropped?
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200660628
--- Diff: flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroRowFormatFactory.java ---
@@ -0,0 +1,97 @@
+/*
+ * 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.formats.avro;
+
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.table.descriptors.AvroValidator;
+import org.apache.flink.table.descriptors.DescriptorProperties;
+import org.apache.flink.table.descriptors.FormatDescriptorValidator;
+import org.apache.flink.table.formats.DeserializationSchemaFactory;
+import org.apache.flink.table.formats.SerializationSchemaFactory;
+import org.apache.flink.types.Row;
+
+import org.apache.avro.specific.SpecificRecord;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Table format factory for providing configured instances of Avro-to-row {@link SerializationSchema}
+ * and {@link DeserializationSchema}.
+ */
+public class AvroRowFormatFactory implements SerializationSchemaFactory<Row>, DeserializationSchemaFactory<Row> {
+
+ @Override
+ public Map<String, String> requiredContext() {
+ final Map<String, String> context = new HashMap<>();
+ context.put(FormatDescriptorValidator.FORMAT_TYPE(), AvroValidator.FORMAT_TYPE_VALUE);
+ context.put(FormatDescriptorValidator.FORMAT_PROPERTY_VERSION(), "1");
+ return context;
+ }
+
+ @Override
+ public boolean supportsSchemaDerivation() {
+ return false;
+ }
+
+ @Override
+ public List<String> supportedProperties() {
+ final List<String> properties = new ArrayList<>();
+ properties.add(AvroValidator.FORMAT_RECORD_CLASS);
+ properties.add(AvroValidator.FORMAT_AVRO_SCHEMA);
+ return properties;
+ }
+
+ @Override
+ public DeserializationSchema<Row> createDeserializationSchema(Map<String, String> properties) {
+ final DescriptorProperties props = new DescriptorProperties(true);
--- End diff --
deduplicate code for creation of `props` and validation
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r201363167
--- Diff: flink-libraries/flink-table/src/test/scala/org/apache/flink/table/formats/TableFormatFactoryServiceTest.scala ---
@@ -0,0 +1,122 @@
+/*
+ * 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.formats
+
+import java.util.{HashMap => JHashMap, Map => JMap}
+
+import org.apache.flink.table.api.{AmbiguousTableFormatException, NoMatchingTableFormatException}
+import org.apache.flink.table.formats.utils.TestAmbiguousTableFormatFactory
+import org.junit.Assert.{assertNotNull, assertTrue}
+import org.junit.Test
+
+/**
+ * Tests for [[TableFormatFactoryService]].
+ */
+class TableFormatFactoryServiceTest {
+
+ @Test
+ def testValidProperties(): Unit = {
+ val props = properties()
+ assertNotNull(
+ TableFormatFactoryService.find(classOf[TableFormatFactory[_]], props))
+ }
+
+ @Test
+ def testDifferentContextVersion(): Unit = {
+ val props = properties()
+ props.put("format.property-version", "2")
+ // the format should still be found
+ assertNotNull(TableFormatFactoryService.find(classOf[TableFormatFactory[_]], props))
+ }
+
+ @Test
+ def testAmbiguousMoreSupportSelection(): Unit = {
+ val props = properties()
+ props.remove("format.important")
+ props.put("format.special_path", "/what/ever")
+ assertTrue(
+ TableFormatFactoryService
+ .find(classOf[TableFormatFactory[_]], props)
+ .isInstanceOf[TestAmbiguousTableFormatFactory])
+ }
+
+ @Test
+ def testAmbiguousClassBasedSelection(): Unit = {
+ val props = properties()
+ props.remove("format.important")
+ assertTrue(
+ TableFormatFactoryService
+ .find(classOf[TestAmbiguousTableFormatFactory], props)
+ .isInstanceOf[TestAmbiguousTableFormatFactory])
+ }
+
+ @Test
+ def testAmbiguousSchemaBasedSelection(): Unit = {
+ val props = properties()
+ props.remove("format.important")
+ props.put("schema.weird_field", "unknown") // this is unknown to the schema derivation factory
+ assertTrue(
+ TableFormatFactoryService
+ .find(classOf[TableFormatFactory[_]], props)
+ .isInstanceOf[TestAmbiguousTableFormatFactory])
+ }
+
+ @Test(expected = classOf[NoMatchingTableFormatException])
+ def testMissingClass(): Unit = {
+ val props = properties()
+ // this class is not a valid factory
+ TableFormatFactoryService.find(classOf[TableFormatFactoryServiceTest], props)
+ }
+
+ @Test(expected = classOf[NoMatchingTableFormatException])
+ def testInvalidContext(): Unit = {
+ val props = properties()
+ props.put("format.type", "FAIL") // no context specifies this
+ TableFormatFactoryService.find(classOf[TableFormatFactory[_]], props)
+ }
+
+ @Test(expected = classOf[NoMatchingTableFormatException])
+ def testUnsupportedProperty(): Unit = {
+ val props = properties()
+ props.put("format.path_new", "/new/path") // no factory has this
+ TableFormatFactoryService.find(classOf[TableFormatFactory[_]], props)
+ }
+
+ @Test(expected = classOf[AmbiguousTableFormatException])
+ def testAmbiguousFactory(): Unit = {
--- End diff --
I added a comment to the other test.
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200999798
--- Diff: flink-libraries/flink-table/src/test/scala/org/apache/flink/table/descriptors/TableSourceDescriptorTest.scala ---
@@ -0,0 +1,133 @@
+/*
+ * 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.descriptors
+
+import org.apache.flink.table.api.Types
+import org.apache.flink.table.utils.TableTestBase
+import org.junit.Assert.assertEquals
+import org.junit.Test
+
+import scala.collection.JavaConverters._
+
+class TableSourceDescriptorTest extends TableTestBase {
+
+ @Test
+ def testStreamTableSourceDescriptor(): Unit = {
+ val util = streamTestUtil()
+ val desc = util.tableEnv
+ .from(
+ FileSystem()
+ .path("/path/to/csv"))
+ .withFormat(
+ Csv()
+ .field("myfield", Types.STRING)
+ .field("myfield2", Types.INT)
+ .quoteCharacter(';')
+ .fieldDelimiter("#")
+ .lineDelimiter("\r\n")
+ .commentPrefix("%%")
+ .ignoreFirstLine()
+ .ignoreParseErrors())
+ .withSchema(
+ Schema()
+ .field("myfield", Types.STRING)
+ .field("myfield2", Types.INT)
+ .field("proctime", Types.SQL_TIMESTAMP).proctime()
+ )
+ val expected = Seq(
+ "connector.property-version" -> "1",
+ "connector.type" -> "filesystem",
+ "connector.path" -> "/path/to/csv",
+ "format.property-version" -> "1",
+ "format.type" -> "csv",
+ "format.fields.0.name" -> "myfield",
+ "format.fields.0.type" -> "VARCHAR",
+ "format.fields.1.name" -> "myfield2",
+ "format.fields.1.type" -> "INT",
+ "format.quote-character" -> ";",
+ "format.field-delimiter" -> "#",
+ "format.line-delimiter" -> "\r\n",
+ "format.comment-prefix" -> "%%",
+ "format.ignore-first-line" -> "true",
+ "format.ignore-parse-errors" -> "true",
+ "schema.0.name" -> "myfield",
+ "schema.0.type" -> "VARCHAR",
+ "schema.1.name" -> "myfield2",
+ "schema.1.type" -> "INT",
+ "schema.2.name" -> "proctime",
+ "schema.2.type" -> "TIMESTAMP",
+ "schema.2.proctime" -> "true"
+ ).toMap
+
+ val actual = new DescriptorProperties(true)
+ desc.addProperties(actual)
+
+ assertEquals(expected.asJava, actual.asMap)
+ }
+
+ @Test
+ def testBatchTableSourceDescriptor(): Unit = {
+ val util = batchTestUtil()
+ val desc = util.tableEnv
--- End diff --
ditto
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200666957
--- Diff: flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonSchemaConverterTest.java ---
@@ -31,7 +31,7 @@
import static org.junit.Assert.assertEquals;
/**
- * Tests for {@link JsonSchemaConverter}.
+ * Tests for {@link JsonRowSchemaConverter}.
--- End diff --
You forgot to rename the test file.
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200707458
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/BatchTableEnvironment.scala ---
@@ -110,16 +111,44 @@ abstract class BatchTableEnvironment(
}
}
-// TODO expose this once we have enough table source factories that can deal with it
-// /**
-// * Creates a table from a descriptor that describes the source connector, source encoding,
-// * the resulting table schema, and other properties.
-// *
-// * @param connectorDescriptor connector descriptor describing the source of the table
-// */
-// def from(connectorDescriptor: ConnectorDescriptor): BatchTableSourceDescriptor = {
-// new BatchTableSourceDescriptor(this, connectorDescriptor)
-// }
+ /**
+ * Creates a table from a descriptor that describes the source connector, the source format,
+ * the resulting table schema, and other properties.
+ *
+ * Descriptors allow for declaring communication to external systems in an
+ * implementation-agnostic way. The classpath is scanned for connectors and matching connectors
+ * are configured accordingly.
+ *
+ * The following example shows how to read from a Kafka connector using a JSON format and
+ * creating a table:
+ *
+ * {{{
+ *
+ * tableEnv
+ * .from(
+ * new Kafka()
+ * .version("0.11")
+ * .topic("clicks")
+ * .property("zookeeper.connect", "localhost")
+ * .property("group.id", "click-group")
+ * .startFromEarliest())
+ * .withFormat(
+ * new Json()
+ * .jsonSchema("{...}")
+ * .failOnMissingField(false))
+ * .withSchema(
+ * new Schema()
+ * .field("user-name", "VARCHAR").from("u_name")
+ * .field("count", "DECIMAL")
+ * .field("proc-time", "TIMESTAMP").proctime())
+ * .toTable()
+ * }}}
+ *
+ * @param connectorDescriptor connector descriptor describing the source of the table
+ */
+ def from(connectorDescriptor: ConnectorDescriptor): BatchTableSourceDescriptor = {
--- End diff --
Do we really want to forward a user to an internal class? I mean actually an IDE should inherit the JavaDoc from the upper class but it might be useful in API classes nevertheless.
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200991456
--- Diff: flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/AvroRowFormatFactoryTest.java ---
@@ -0,0 +1,91 @@
+/*
+ * 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.formats.avro;
+
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.formats.avro.generated.User;
+import org.apache.flink.table.descriptors.Avro;
+import org.apache.flink.table.descriptors.Descriptor;
+import org.apache.flink.table.descriptors.DescriptorProperties;
+import org.apache.flink.table.formats.DeserializationSchemaFactory;
+import org.apache.flink.table.formats.SerializationSchemaFactory;
+import org.apache.flink.table.formats.TableFormatFactoryService;
+
+import org.junit.Test;
+
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Tests for the {@link AvroRowFormatFactory}.
+ */
+public class AvroRowFormatFactoryTest {
--- End diff --
Hmm, @tillrohrmann was always bugging me about extending from `TestLogger` since it helps in case of deadlocks. However I might be missing something?
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr closed the pull request at:
https://github.com/apache/flink/pull/6264
---
[GitHub] flink issue #6264: [FLINK-8558] [table] Add unified format interfaces and se...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on the issue:
https://github.com/apache/flink/pull/6264
Thanks for the in-depth review @pnowojski. I hope I could address most of your comments. Since this PR heavily overlaps with #6201 and that PR needs a review and some additional work as well. I will close this PR for now and open a PR with a clean unified table sources/sinks/formats story. We can continue the discussions here and I will make sure that changes will be applied to new PR as well.
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r201019023
--- Diff: flink-libraries/flink-table/src/test/scala/org/apache/flink/table/formats/TableFormatFactoryServiceTest.scala ---
@@ -0,0 +1,122 @@
+/*
+ * 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.formats
+
+import java.util.{HashMap => JHashMap, Map => JMap}
+
+import org.apache.flink.table.api.{AmbiguousTableFormatException, NoMatchingTableFormatException}
+import org.apache.flink.table.formats.utils.TestAmbiguousTableFormatFactory
+import org.junit.Assert.{assertNotNull, assertTrue}
+import org.junit.Test
+
+/**
+ * Tests for [[TableFormatFactoryService]].
+ */
+class TableFormatFactoryServiceTest {
+
+ @Test
+ def testValidProperties(): Unit = {
+ val props = properties()
+ assertNotNull(
+ TableFormatFactoryService.find(classOf[TableFormatFactory[_]], props))
+ }
+
+ @Test
+ def testDifferentContextVersion(): Unit = {
+ val props = properties()
+ props.put("format.property-version", "2")
+ // the format should still be found
+ assertNotNull(TableFormatFactoryService.find(classOf[TableFormatFactory[_]], props))
+ }
+
+ @Test
+ def testAmbiguousMoreSupportSelection(): Unit = {
+ val props = properties()
+ props.remove("format.important")
+ props.put("format.special_path", "/what/ever")
--- End diff --
How does this test differs from `testAmbiguousClassBasedSelection`? What does it test?
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200998068
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/formats/TableFormatFactoryService.scala ---
@@ -0,0 +1,245 @@
+/*
+ * 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.formats
+
+import java.util.{ServiceConfigurationError, ServiceLoader, Map => JMap}
+
+import org.apache.flink.table.api._
+import org.apache.flink.table.descriptors.FormatDescriptorValidator.FORMAT_PROPERTY_VERSION
+import org.apache.flink.table.descriptors._
+import org.apache.flink.table.util.Logging
+
+import _root_.scala.collection.JavaConverters._
+import _root_.scala.collection.mutable
+
+/**
+ * Service provider interface for finding a suitable [[TableFormatFactory]] for the
+ * given properties.
+ */
+object TableFormatFactoryService extends Logging {
+
+ private lazy val defaultLoader = ServiceLoader.load(classOf[TableFormatFactory[_]])
+
+ /**
+ * Finds a table format factory of the given class and creates configured instances from the
+ * given descriptor.
+ *
+ * @param factoryClass desired format factory
+ * @param descriptor descriptor that describes the format
+ * @tparam T factory class type
+ * @return configured instance from factory
+ */
+ def find[T](factoryClass: Class[T], descriptor: Descriptor): T = {
+ find(factoryClass, descriptor, null)
+ }
+
+ /**
+ * Finds a table format factory of the given class and creates configured instances from the
+ * given descriptor and classloader.
+ *
+ * @param factoryClass desired format factory
+ * @param descriptor descriptor that describes the format
+ * @param classLoader classloader for service loading
+ * @tparam T factory class type
+ * @return configured instance from factory
+ */
+ def find[T](
+ factoryClass: Class[T],
+ descriptor: Descriptor,
+ classLoader: ClassLoader)
+ : T = {
+
+ val properties = new DescriptorProperties()
+ descriptor.addProperties(properties)
+ find(factoryClass, properties.asMap, classLoader)
+ }
+
+ /**
+ * Finds a table format factory of the given class and creates configured instances from the
+ * given property map.
+ *
+ * @param factoryClass desired format factory
+ * @param propertyMap properties that describes the format
+ * @tparam T factory class type
+ * @return configured instance from factory
+ */
+ def find[T](factoryClass: Class[T], propertyMap: JMap[String, String]): T = {
+ find(factoryClass, propertyMap, null)
+ }
+
+ /**
+ * Finds a table format factory of the given class and creates configured instances from the
+ * given property map and classloader.
+ *
+ * @param factoryClass desired format factory
+ * @param propertyMap properties that describes the format
+ * @param classLoader classloader for service loading
+ * @tparam T factory class type
+ * @return configured instance from factory
+ */
+ def find[T](
+ factoryClass: Class[T],
+ propertyMap: JMap[String, String],
+ classLoader: ClassLoader)
+ : T = {
+
+ val properties = propertyMap.asScala.toMap
+
+ // find matching context
+ val (foundFactories, contextFactories) = findMatchingContext(properties, classLoader)
+ if (contextFactories.isEmpty) {
+ throw new NoMatchingTableFormatException(
+ "No context matches.",
+ factoryClass,
+ foundFactories,
+ properties)
+ }
+
+ // filter for factory class
+ val classFactories = contextFactories.filter(f => factoryClass.isAssignableFrom(f.getClass))
+ if (classFactories.isEmpty) {
+ throw new NoMatchingTableFormatException(
+ s"No factory implements '${factoryClass.getCanonicalName}'.",
+ factoryClass,
+ foundFactories,
+ properties)
+ }
+
+ // filter by supported keys
+ val plainGivenKeys = mutable.ArrayBuffer[String]()
+ properties.keys.foreach { k =>
+ // replace arrays with wildcard
+ val key = k.replaceAll(".\\d+", ".#")
+ // ignore duplicates
+ if (!plainGivenKeys.contains(key)) {
+ plainGivenKeys += key
+ }
+ }
+ var lastKey: Option[String] = None
+ val supportedFactories = classFactories.filter { factory =>
+ val requiredContextKeys = normalizeContext(factory).keySet
+ val includeSchema = factory.supportsSchemaDerivation()
+ val supportedKeys = normalizeSupportedProperties(factory)
+ val givenKeys = plainGivenKeys
+ // ignore context keys
+ .filter(!requiredContextKeys.contains(_))
+ // ignore non-format (or schema) keys
+ .filter { k =>
+ if (includeSchema) {
+ k.startsWith(SchemaValidator.SCHEMA + ".") ||
+ k.startsWith(FormatDescriptorValidator.FORMAT + ".")
+ } else {
+ k.startsWith(FormatDescriptorValidator.FORMAT + ".")
+ }
+ }
+ givenKeys.forall { k =>
+ lastKey = Option(k)
+ supportedKeys.contains(k)
+ }
+ }
+ if (supportedFactories.isEmpty && classFactories.length == 1 && lastKey.isDefined) {
+ // special case: when there is only one matching factory but the last property key
+ // was incorrect
+ val factory = classFactories.head
+ val supportedKeys = normalizeSupportedProperties(factory)
+ throw new NoMatchingTableFormatException(
+ s"""
+ |The matching factory '${factory.getClass.getName}' doesn't support '${lastKey.get}'.
+ |
+ |Supported properties of this factory are:
+ |${supportedKeys.sorted.mkString("\n")}""".stripMargin,
+ factoryClass,
+ foundFactories,
+ properties)
+ } else if (supportedFactories.isEmpty) {
+ throw new NoMatchingTableFormatException(
+ s"No factory supports all properties.",
+ factoryClass,
+ foundFactories,
+ properties)
+ } else if (supportedFactories.length > 1) {
+ throw new AmbiguousTableFormatException(
+ supportedFactories,
+ factoryClass,
+ foundFactories,
+ properties)
+ }
+
+ supportedFactories.head.asInstanceOf[T]
+ }
+
+ private def findMatchingContext(
+ properties: Map[String, String],
+ classLoader: ClassLoader)
+ : (Seq[TableFormatFactory[_]], Seq[TableFormatFactory[_]]) = {
+
+ val foundFactories = mutable.ArrayBuffer[TableFormatFactory[_]]()
+ val matchingFactories = mutable.ArrayBuffer[TableFormatFactory[_]]()
+
+ try {
+ val iter = if (classLoader == null) {
+ defaultLoader.iterator()
+ } else {
+ val customLoader = ServiceLoader.load(classOf[TableFormatFactory[_]], classLoader)
+ customLoader.iterator()
+ }
+ while (iter.hasNext) {
+ val factory = iter.next()
+ foundFactories += factory
+
+ val requestedContext = normalizeContext(factory)
+
+ val plainContext = mutable.Map[String, String]()
+ plainContext ++= requestedContext
+ // we remove the version for now until we have the first backwards compatibility case
+ // with the version we can provide mappings in case the format changes
+ plainContext.remove(FORMAT_PROPERTY_VERSION)
+
+ // check if required context is met
+ if (plainContext.forall(e => properties.contains(e._1) && properties(e._1) == e._2)) {
+ matchingFactories += factory
+ }
+ }
+ } catch {
+ case e: ServiceConfigurationError =>
+ LOG.error("Could not load service provider for table format factories.", e)
+ throw new TableException("Could not load service provider for table format factories.", e)
+ }
+
+ (foundFactories, matchingFactories)
+ }
+
+ private def normalizeContext(factory: TableFormatFactory[_]): Map[String, String] = {
+ val requiredContextJava = factory.requiredContext()
+ if (requiredContextJava != null) {
+ requiredContextJava.asScala.map(e => (e._1.toLowerCase, e._2)).toMap
+ } else {
+ Map[String, String]()
+ }
+ }
+
+ private def normalizeSupportedProperties(factory: TableFormatFactory[_]): Seq[String] = {
+ val supportedPropertiesJava = factory.supportedProperties()
+ if (supportedPropertiesJava != null) {
--- End diff --
ok, in that case why not use `checkNotNull()`?
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r201353196
--- Diff: flink-libraries/flink-table/src/test/scala/org/apache/flink/table/formats/TableFormatFactoryServiceTest.scala ---
@@ -0,0 +1,122 @@
+/*
+ * 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.formats
+
+import java.util.{HashMap => JHashMap, Map => JMap}
+
+import org.apache.flink.table.api.{AmbiguousTableFormatException, NoMatchingTableFormatException}
+import org.apache.flink.table.formats.utils.TestAmbiguousTableFormatFactory
+import org.junit.Assert.{assertNotNull, assertTrue}
+import org.junit.Test
+
+/**
+ * Tests for [[TableFormatFactoryService]].
+ */
+class TableFormatFactoryServiceTest {
+
+ @Test
+ def testValidProperties(): Unit = {
+ val props = properties()
+ assertNotNull(
+ TableFormatFactoryService.find(classOf[TableFormatFactory[_]], props))
+ }
+
+ @Test
+ def testDifferentContextVersion(): Unit = {
+ val props = properties()
+ props.put("format.property-version", "2")
+ // the format should still be found
+ assertNotNull(TableFormatFactoryService.find(classOf[TableFormatFactory[_]], props))
+ }
+
+ @Test
+ def testAmbiguousMoreSupportSelection(): Unit = {
+ val props = properties()
+ props.remove("format.important")
+ props.put("format.special_path", "/what/ever")
--- End diff --
Sorry, I should have added more comments to these tests. Will do this now...
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r201026527
--- Diff: flink-libraries/flink-table/src/test/scala/org/apache/flink/table/formats/TableFormatFactoryServiceTest.scala ---
@@ -0,0 +1,122 @@
+/*
+ * 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.formats
+
+import java.util.{HashMap => JHashMap, Map => JMap}
+
+import org.apache.flink.table.api.{AmbiguousTableFormatException, NoMatchingTableFormatException}
+import org.apache.flink.table.formats.utils.TestAmbiguousTableFormatFactory
+import org.junit.Assert.{assertNotNull, assertTrue}
+import org.junit.Test
+
+/**
+ * Tests for [[TableFormatFactoryService]].
+ */
+class TableFormatFactoryServiceTest {
+
+ @Test
+ def testValidProperties(): Unit = {
+ val props = properties()
+ assertNotNull(
+ TableFormatFactoryService.find(classOf[TableFormatFactory[_]], props))
+ }
+
+ @Test
+ def testDifferentContextVersion(): Unit = {
+ val props = properties()
+ props.put("format.property-version", "2")
+ // the format should still be found
+ assertNotNull(TableFormatFactoryService.find(classOf[TableFormatFactory[_]], props))
+ }
+
+ @Test
+ def testAmbiguousMoreSupportSelection(): Unit = {
+ val props = properties()
+ props.remove("format.important")
+ props.put("format.special_path", "/what/ever")
+ assertTrue(
+ TableFormatFactoryService
+ .find(classOf[TableFormatFactory[_]], props)
+ .isInstanceOf[TestAmbiguousTableFormatFactory])
+ }
+
+ @Test
+ def testAmbiguousClassBasedSelection(): Unit = {
+ val props = properties()
+ props.remove("format.important")
+ assertTrue(
+ TableFormatFactoryService
+ .find(classOf[TestAmbiguousTableFormatFactory], props)
+ .isInstanceOf[TestAmbiguousTableFormatFactory])
+ }
+
+ @Test
+ def testAmbiguousSchemaBasedSelection(): Unit = {
+ val props = properties()
+ props.remove("format.important")
+ props.put("schema.weird_field", "unknown") // this is unknown to the schema derivation factory
+ assertTrue(
+ TableFormatFactoryService
+ .find(classOf[TableFormatFactory[_]], props)
+ .isInstanceOf[TestAmbiguousTableFormatFactory])
+ }
+
+ @Test(expected = classOf[NoMatchingTableFormatException])
+ def testMissingClass(): Unit = {
+ val props = properties()
+ // this class is not a valid factory
+ TableFormatFactoryService.find(classOf[TableFormatFactoryServiceTest], props)
+ }
+
+ @Test(expected = classOf[NoMatchingTableFormatException])
+ def testInvalidContext(): Unit = {
+ val props = properties()
+ props.put("format.type", "FAIL") // no context specifies this
+ TableFormatFactoryService.find(classOf[TableFormatFactory[_]], props)
+ }
+
+ @Test(expected = classOf[NoMatchingTableFormatException])
+ def testUnsupportedProperty(): Unit = {
+ val props = properties()
+ props.put("format.path_new", "/new/path") // no factory has this
--- End diff --
`s/path_new/property_not_defined_by_any_factory/g` and remove comment
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r201001341
--- Diff: flink-libraries/flink-table/src/test/scala/org/apache/flink/table/descriptors/TableSourceDescriptorTest.scala ---
@@ -0,0 +1,133 @@
+/*
+ * 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.descriptors
+
+import org.apache.flink.table.api.Types
+import org.apache.flink.table.utils.TableTestBase
+import org.junit.Assert.assertEquals
+import org.junit.Test
+
+import scala.collection.JavaConverters._
+
+class TableSourceDescriptorTest extends TableTestBase {
+
+ @Test
+ def testStreamTableSourceDescriptor(): Unit = {
+ val util = streamTestUtil()
+ val desc = util.tableEnv
+ .from(
+ FileSystem()
+ .path("/path/to/csv"))
+ .withFormat(
+ Csv()
+ .field("myfield", Types.STRING)
+ .field("myfield2", Types.INT)
+ .quoteCharacter(';')
+ .fieldDelimiter("#")
+ .lineDelimiter("\r\n")
+ .commentPrefix("%%")
+ .ignoreFirstLine()
+ .ignoreParseErrors())
+ .withSchema(
+ Schema()
+ .field("myfield", Types.STRING)
+ .field("myfield2", Types.INT)
+ .field("proctime", Types.SQL_TIMESTAMP).proctime()
+ )
+ val expected = Seq(
+ "connector.property-version" -> "1",
+ "connector.type" -> "filesystem",
+ "connector.path" -> "/path/to/csv",
+ "format.property-version" -> "1",
+ "format.type" -> "csv",
+ "format.fields.0.name" -> "myfield",
+ "format.fields.0.type" -> "VARCHAR",
+ "format.fields.1.name" -> "myfield2",
+ "format.fields.1.type" -> "INT",
+ "format.quote-character" -> ";",
+ "format.field-delimiter" -> "#",
+ "format.line-delimiter" -> "\r\n",
+ "format.comment-prefix" -> "%%",
+ "format.ignore-first-line" -> "true",
+ "format.ignore-parse-errors" -> "true",
+ "schema.0.name" -> "myfield",
+ "schema.0.type" -> "VARCHAR",
+ "schema.1.name" -> "myfield2",
+ "schema.1.type" -> "INT",
+ "schema.2.name" -> "proctime",
+ "schema.2.type" -> "TIMESTAMP",
+ "schema.2.proctime" -> "true"
+ ).toMap
+
+ val actual = new DescriptorProperties(true)
+ desc.addProperties(actual)
+
+ assertEquals(expected.asJava, actual.asMap)
+ }
+
+ @Test
+ def testBatchTableSourceDescriptor(): Unit = {
--- End diff --
please deduplicate this with streaming case. It seems like they differ only on used `util` type (that have common subtype) and one field.
Thus you could create a common private method
```
@Test
def testBatchTableSourceDescriptor = testTableSourceDescriptor(false)
@Test
def testStreamingTableSourceDescriptor = testTableSourceDescriptor(true)
def testTableSourceDescriptor(isStreaming: Boolean) {
// deduplicated code here with 3 small if's
}
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200569545
--- Diff: flink-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010TableSource.java ---
@@ -18,48 +18,99 @@
package org.apache.flink.streaming.connectors.kafka;
-import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.streaming.connectors.kafka.config.StartupMode;
+import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.sources.RowtimeAttributeDescriptor;
import org.apache.flink.table.sources.StreamTableSource;
import org.apache.flink.types.Row;
+import java.util.List;
+import java.util.Map;
import java.util.Properties;
/**
* Kafka {@link StreamTableSource} for Kafka 0.10.
*/
-@PublicEvolving
-public abstract class Kafka010TableSource extends KafkaTableSource {
+@Internal
+public class Kafka010TableSource extends KafkaTableSource {
- // The deserialization schema for the Kafka records
- private final DeserializationSchema<Row> deserializationSchema;
+ /**
+ * Creates a Kafka 0.10 {@link StreamTableSource}.
+ *
+ * @param schema Schema of the produced table.
+ * @param proctimeAttribute Field name of the processing time attribute, null if no
+ * processing time field is defined.
+ * @param rowtimeAttributeDescriptors Descriptor for a rowtime attribute
+ * @param fieldMapping Mapping for the fields of the table schema to
+ * fields of the physical returned type or null.
+ * @param topic Kafka topic to consume.
+ * @param properties Properties for the Kafka consumer.
+ * @param deserializationSchema Deserialization schema for decoding records from Kafka.
+ * @param startupMode Startup mode for the contained consumer.
+ * @param specificStartupOffsets Specific startup offsets; only relevant when startup
+ * mode is {@link StartupMode#SPECIFIC_OFFSETS}.
+ */
+ public Kafka010TableSource(
+ TableSchema schema,
+ String proctimeAttribute,
+ List<RowtimeAttributeDescriptor> rowtimeAttributeDescriptors,
+ Map<String, String> fieldMapping,
+ String topic, Properties properties,
--- End diff --
nit: new line between `topic` & `properties`
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200694738
--- Diff: flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceFactoryTestBase.java ---
@@ -0,0 +1,175 @@
+/*
+ * 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.streaming.connectors.kafka;
+
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.streaming.connectors.kafka.config.StartupMode;
+import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.api.Types;
+import org.apache.flink.table.descriptors.Kafka;
+import org.apache.flink.table.descriptors.Rowtime;
+import org.apache.flink.table.descriptors.Schema;
+import org.apache.flink.table.descriptors.TestTableSourceDescriptor;
+import org.apache.flink.table.formats.utils.TestDeserializationSchema;
+import org.apache.flink.table.formats.utils.TestTableFormat;
+import org.apache.flink.table.sources.RowtimeAttributeDescriptor;
+import org.apache.flink.table.sources.TableSource;
+import org.apache.flink.table.sources.TableSourceFactoryService;
+import org.apache.flink.table.sources.tsextractors.ExistingField;
+import org.apache.flink.table.sources.wmstrategies.AscendingTimestamps;
+import org.apache.flink.types.Row;
+
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Abstract test base for {@link KafkaTableSourceFactory}.
+ */
+public abstract class KafkaTableSourceFactoryTestBase {
--- End diff --
The value of this test is to test the entire translation logic end-to-end. We have tests for each descriptor and each factory but we also need to make sure that all the components work together and the final result is the desired one.
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200908582
--- Diff: flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSource.java ---
@@ -55,50 +57,101 @@
*/
@Internal
public abstract class KafkaTableSource
- implements StreamTableSource<Row>, DefinedProctimeAttribute, DefinedRowtimeAttributes {
+ implements StreamTableSource<Row>, DefinedProctimeAttribute, DefinedRowtimeAttributes, DefinedFieldMapping {
+
+ // common table source attributes
+ // TODO make all attributes final once we drop support for format-specific table sources
/** The schema of the table. */
private final TableSchema schema;
+ /** Field name of the processing time attribute, null if no processing time field is defined. */
+ private String proctimeAttribute;
+
+ /** Descriptor for a rowtime attribute. */
+ private List<RowtimeAttributeDescriptor> rowtimeAttributeDescriptors;
+
+ /** Mapping for the fields of the table schema to fields of the physical returned type or null. */
+ private Map<String, String> fieldMapping;
--- End diff --
I added a `@Nullable` annotation.
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r201611503
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/formats/TableFormatFactory.scala ---
@@ -0,0 +1,85 @@
+/*
+ * 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.formats
+
+import java.util
+
+import org.apache.flink.api.common.serialization.{DeserializationSchema, SerializationSchema}
+
+/**
+ * A factory to create different table format instances. This factory is used with Java's Service
+ * Provider Interfaces (SPI) for discovering. A factory is called with a set of normalized
+ * properties that describe the desired format. The factory allows for matching to the given set of
+ * properties. See also [[SerializationSchemaFactory]] and [[DeserializationSchemaFactory]] for
+ * creating configured instances of format classes accordingly.
+ *
+ * Classes that implement this interface need to be added to the
+ * "META_INF/services/org.apache.flink.table.formats.TableFormatFactory' file of a JAR file in
+ * the current classpath to be found.
+ *
+ * @tparam T record type that the format produces or consumes
+ */
+trait TableFormatFactory[T] {
+
+ /**
+ * Specifies the context that this factory has been implemented for. The framework guarantees
+ * to only use the factory if the specified set of properties and values are met.
+ *
+ * Typical properties might be:
+ * - format.type
+ * - format.version
+ *
+ * Specified property versions allow the framework to provide backwards compatible properties
+ * in case of string format changes:
+ * - format.property-version
+ *
+ * An empty context means that the factory matches for all requests.
+ */
+ def requiredContext(): util.Map[String, String]
--- End diff --
Context is more accurate as the map also contains properties that are not required. E.g. a user does not have to specify `property-version` in YAML but he can. The required context explains the context for which this factory was implemented for.
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r201345819
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/formats/TableFormatFactoryService.scala ---
@@ -0,0 +1,245 @@
+/*
+ * 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.formats
+
+import java.util.{ServiceConfigurationError, ServiceLoader, Map => JMap}
+
+import org.apache.flink.table.api._
+import org.apache.flink.table.descriptors.FormatDescriptorValidator.FORMAT_PROPERTY_VERSION
+import org.apache.flink.table.descriptors._
+import org.apache.flink.table.util.Logging
+
+import _root_.scala.collection.JavaConverters._
+import _root_.scala.collection.mutable
+
+/**
+ * Service provider interface for finding a suitable [[TableFormatFactory]] for the
+ * given properties.
+ */
+object TableFormatFactoryService extends Logging {
--- End diff --
This class is reworked for unified sinks anyway. I will make sure to clean up the duplicate code.
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200666882
--- Diff: flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonSchemaConverterTest.java ---
@@ -31,7 +31,7 @@
import static org.junit.Assert.assertEquals;
/**
- * Tests for {@link JsonSchemaConverter}.
+ * Tests for {@link JsonRowSchemaConverter}.
--- End diff --
rename in separate commit :(
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r201351507
--- Diff: flink-libraries/flink-table/src/test/scala/org/apache/flink/table/formats/TableFormatFactoryServiceTest.scala ---
@@ -0,0 +1,122 @@
+/*
+ * 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.formats
+
+import java.util.{HashMap => JHashMap, Map => JMap}
+
+import org.apache.flink.table.api.{AmbiguousTableFormatException, NoMatchingTableFormatException}
+import org.apache.flink.table.formats.utils.TestAmbiguousTableFormatFactory
+import org.junit.Assert.{assertNotNull, assertTrue}
+import org.junit.Test
+
+/**
+ * Tests for [[TableFormatFactoryService]].
+ */
+class TableFormatFactoryServiceTest {
+
+ @Test
+ def testValidProperties(): Unit = {
+ val props = properties()
+ assertNotNull(
+ TableFormatFactoryService.find(classOf[TableFormatFactory[_]], props))
+ }
+
+ @Test
+ def testDifferentContextVersion(): Unit = {
+ val props = properties()
+ props.put("format.property-version", "2")
--- End diff --
For now we support any property version. The property version should not affect the discovery at the moment. I added a comment to the test.
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200696183
--- Diff: flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroRowFormatFactory.java ---
@@ -0,0 +1,97 @@
+/*
+ * 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.formats.avro;
+
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.table.descriptors.AvroValidator;
+import org.apache.flink.table.descriptors.DescriptorProperties;
+import org.apache.flink.table.descriptors.FormatDescriptorValidator;
+import org.apache.flink.table.formats.DeserializationSchemaFactory;
+import org.apache.flink.table.formats.SerializationSchemaFactory;
+import org.apache.flink.types.Row;
+
+import org.apache.avro.specific.SpecificRecord;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Table format factory for providing configured instances of Avro-to-row {@link SerializationSchema}
+ * and {@link DeserializationSchema}.
+ */
+public class AvroRowFormatFactory implements SerializationSchemaFactory<Row>, DeserializationSchemaFactory<Row> {
+
+ @Override
+ public Map<String, String> requiredContext() {
+ final Map<String, String> context = new HashMap<>();
+ context.put(FormatDescriptorValidator.FORMAT_TYPE(), AvroValidator.FORMAT_TYPE_VALUE);
+ context.put(FormatDescriptorValidator.FORMAT_PROPERTY_VERSION(), "1");
+ return context;
+ }
+
+ @Override
+ public boolean supportsSchemaDerivation() {
+ return false;
+ }
+
+ @Override
+ public List<String> supportedProperties() {
+ final List<String> properties = new ArrayList<>();
+ properties.add(AvroValidator.FORMAT_RECORD_CLASS);
+ properties.add(AvroValidator.FORMAT_AVRO_SCHEMA);
+ return properties;
+ }
+
+ @Override
+ public DeserializationSchema<Row> createDeserializationSchema(Map<String, String> properties) {
+ final DescriptorProperties props = new DescriptorProperties(true);
--- End diff --
Validation and creation are independent tasks. Validation is an internal task that's why validators have internal utility classes (`DescriptorProperties`), however, the public API for users that want to implement own formats have the plain-old `java.util.Map`.
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200597649
--- Diff: flink-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010AvroTableSource.java ---
@@ -46,7 +53,9 @@
* @param properties Properties for the Kafka consumer.
* @param schema Schema of the produced table.
* @param record Avro specific record.
+ * @deprecated Use table descriptors instead of implementation-specific builders.
*/
+ @Deprecated
--- End diff --
Hmm why do you deprecate individual methods after already deprecating whole class? Does it solve some problem?
If not I would revert those additional `@Deprecated` notes.
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200685042
--- Diff: flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceFactory.java ---
@@ -124,110 +123,130 @@
params.putProperties(properties);
// validate
- new SchemaValidator(true).validate(params);
+ // allow Kafka timestamps to be used, watermarks can not be received from source
+ new SchemaValidator(true, supportsKafkaTimestamps(), false).validate(params);
new KafkaValidator().validate(params);
- formatValidator().validate(params);
- // build
- final KafkaTableSource.Builder builder = createBuilderWithFormat(params);
+ // deserialization schema using format discovery
+ final DeserializationSchemaFactory<?> formatFactory = TableFormatFactoryService.find(
+ DeserializationSchemaFactory.class,
+ properties,
+ this.getClass().getClassLoader());
+ @SuppressWarnings("unchecked")
+ final DeserializationSchema<Row> deserializationSchema = (DeserializationSchema<Row>) formatFactory
+ .createDeserializationSchema(properties);
- // topic
- final String topic = params.getString(CONNECTOR_TOPIC);
- builder.forTopic(topic);
+ // schema
--- End diff --
Sorry, about that. I would not only focus on the diff but also have a look at the final result. Because this is what matters in the end. But I will try to split those changes into separate commits in the future.
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200677056
--- Diff: flink-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010TableSource.java ---
@@ -18,48 +18,99 @@
package org.apache.flink.streaming.connectors.kafka;
-import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.streaming.connectors.kafka.config.StartupMode;
+import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.sources.RowtimeAttributeDescriptor;
import org.apache.flink.table.sources.StreamTableSource;
import org.apache.flink.types.Row;
+import java.util.List;
+import java.util.Map;
import java.util.Properties;
/**
* Kafka {@link StreamTableSource} for Kafka 0.10.
*/
-@PublicEvolving
-public abstract class Kafka010TableSource extends KafkaTableSource {
+@Internal
+public class Kafka010TableSource extends KafkaTableSource {
- // The deserialization schema for the Kafka records
- private final DeserializationSchema<Row> deserializationSchema;
+ /**
+ * Creates a Kafka 0.10 {@link StreamTableSource}.
+ *
+ * @param schema Schema of the produced table.
+ * @param proctimeAttribute Field name of the processing time attribute, null if no
+ * processing time field is defined.
+ * @param rowtimeAttributeDescriptors Descriptor for a rowtime attribute
+ * @param fieldMapping Mapping for the fields of the table schema to
+ * fields of the physical returned type or null.
+ * @param topic Kafka topic to consume.
+ * @param properties Properties for the Kafka consumer.
+ * @param deserializationSchema Deserialization schema for decoding records from Kafka.
+ * @param startupMode Startup mode for the contained consumer.
+ * @param specificStartupOffsets Specific startup offsets; only relevant when startup
+ * mode is {@link StartupMode#SPECIFIC_OFFSETS}.
+ */
+ public Kafka010TableSource(
+ TableSchema schema,
+ String proctimeAttribute,
+ List<RowtimeAttributeDescriptor> rowtimeAttributeDescriptors,
+ Map<String, String> fieldMapping,
+ String topic, Properties properties,
+ DeserializationSchema<Row> deserializationSchema,
+ StartupMode startupMode,
+ Map<KafkaTopicPartition, Long> specificStartupOffsets) {
+
+ super(
+ schema,
+ proctimeAttribute,
+ rowtimeAttributeDescriptors,
+ fieldMapping,
+ topic,
+ properties,
+ deserializationSchema,
+ startupMode,
+ specificStartupOffsets);
+ }
+
+ /**
+ * Creates a Kafka 0.10 {@link StreamTableSource}.
+ *
+ * @param schema Schema of the produced table.
+ * @param topic Kafka topic to consume.
+ * @param properties Properties for the Kafka consumer.
+ * @param deserializationSchema Deserialization schema for decoding records from Kafka.
+ */
+ public Kafka010TableSource(
+ TableSchema schema,
+ String topic,
+ Properties properties,
+ DeserializationSchema<Row> deserializationSchema) {
+
+ super(schema, topic, properties, deserializationSchema);
+ }
/**
* Creates a Kafka 0.10 {@link StreamTableSource}.
*
* @param topic Kafka topic to consume.
* @param properties Properties for the Kafka consumer.
* @param deserializationSchema Deserialization schema to use for Kafka records.
- * @param typeInfo Type information describing the result type. The field names are used
- * to parse the JSON file and so are the types.
+ * @param typeInfo Not relevant anymore.
--- End diff --
The `typeInfo` should have been the same as the produced type that has been returned by the `deserializationSchema`. I'm also fine with dropping the constructor already.
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200665932
--- Diff: flink-formats/flink-json/src/test/java/org/apache/flink/formats/json/JsonRowFormatFactoryTest.java ---
@@ -0,0 +1,147 @@
+/*
+ * 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.formats.json;
+
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.table.api.Types;
+import org.apache.flink.table.descriptors.Descriptor;
+import org.apache.flink.table.descriptors.DescriptorProperties;
+import org.apache.flink.table.descriptors.Json;
+import org.apache.flink.table.descriptors.Schema;
+import org.apache.flink.table.formats.DeserializationSchemaFactory;
+import org.apache.flink.table.formats.SerializationSchemaFactory;
+import org.apache.flink.table.formats.TableFormatFactoryService;
+import org.apache.flink.types.Row;
+
+import org.junit.Test;
+
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Tests for the {@link JsonRowFormatFactory}.
+ */
+public class JsonRowFormatFactoryTest {
--- End diff --
Does this test fully test `JsonRowFormatFactory`? What about supported/unsupported properties for example?
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200597054
--- Diff: flink-connectors/flink-connector-kafka-0.10/src/main/java/org/apache/flink/streaming/connectors/kafka/Kafka010TableSource.java ---
@@ -18,48 +18,99 @@
package org.apache.flink.streaming.connectors.kafka;
-import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.streaming.connectors.kafka.config.StartupMode;
+import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.sources.RowtimeAttributeDescriptor;
import org.apache.flink.table.sources.StreamTableSource;
import org.apache.flink.types.Row;
+import java.util.List;
+import java.util.Map;
import java.util.Properties;
/**
* Kafka {@link StreamTableSource} for Kafka 0.10.
*/
-@PublicEvolving
-public abstract class Kafka010TableSource extends KafkaTableSource {
+@Internal
+public class Kafka010TableSource extends KafkaTableSource {
- // The deserialization schema for the Kafka records
- private final DeserializationSchema<Row> deserializationSchema;
+ /**
+ * Creates a Kafka 0.10 {@link StreamTableSource}.
+ *
+ * @param schema Schema of the produced table.
+ * @param proctimeAttribute Field name of the processing time attribute, null if no
+ * processing time field is defined.
+ * @param rowtimeAttributeDescriptors Descriptor for a rowtime attribute
+ * @param fieldMapping Mapping for the fields of the table schema to
+ * fields of the physical returned type or null.
+ * @param topic Kafka topic to consume.
+ * @param properties Properties for the Kafka consumer.
+ * @param deserializationSchema Deserialization schema for decoding records from Kafka.
+ * @param startupMode Startup mode for the contained consumer.
+ * @param specificStartupOffsets Specific startup offsets; only relevant when startup
+ * mode is {@link StartupMode#SPECIFIC_OFFSETS}.
+ */
+ public Kafka010TableSource(
+ TableSchema schema,
+ String proctimeAttribute,
+ List<RowtimeAttributeDescriptor> rowtimeAttributeDescriptors,
+ Map<String, String> fieldMapping,
+ String topic, Properties properties,
+ DeserializationSchema<Row> deserializationSchema,
+ StartupMode startupMode,
+ Map<KafkaTopicPartition, Long> specificStartupOffsets) {
+
+ super(
+ schema,
+ proctimeAttribute,
+ rowtimeAttributeDescriptors,
+ fieldMapping,
+ topic,
+ properties,
+ deserializationSchema,
+ startupMode,
+ specificStartupOffsets);
+ }
+
+ /**
+ * Creates a Kafka 0.10 {@link StreamTableSource}.
+ *
+ * @param schema Schema of the produced table.
+ * @param topic Kafka topic to consume.
+ * @param properties Properties for the Kafka consumer.
+ * @param deserializationSchema Deserialization schema for decoding records from Kafka.
+ */
+ public Kafka010TableSource(
+ TableSchema schema,
+ String topic,
+ Properties properties,
+ DeserializationSchema<Row> deserializationSchema) {
+
+ super(schema, topic, properties, deserializationSchema);
+ }
/**
* Creates a Kafka 0.10 {@link StreamTableSource}.
*
* @param topic Kafka topic to consume.
* @param properties Properties for the Kafka consumer.
* @param deserializationSchema Deserialization schema to use for Kafka records.
- * @param typeInfo Type information describing the result type. The field names are used
- * to parse the JSON file and so are the types.
+ * @param typeInfo Not relevant anymore.
--- End diff --
Why is it not relevant? Doesn't it brake the backward compatibility? If so, it would be safer to drop the constructor altogether.
The only reason for keeping it, would be if **ALL** old invocations will still work the same as they used to, regardless of this value. However in that case, I would be also inclined to drop this constructor, since it's easy change for the users and the class was `@PublicEvolving` and now it's `@Internal`
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r201030536
--- Diff: flink-libraries/flink-table/src/test/scala/org/apache/flink/table/sources/TableSourceFactoryServiceTest.scala ---
@@ -52,21 +57,33 @@ class TableSourceFactoryServiceTest {
@Test(expected = classOf[ValidationException])
def testUnsupportedProperty(): Unit = {
val props = properties()
+ props.put(CONNECTOR_TYPE, "fixed")
+ props.put(FORMAT_TYPE, "test")
props.put("format.path_new", "/new/path")
TableSourceFactoryService.findAndCreateTableSource(props.toMap)
}
@Test(expected = classOf[TableException])
def testFailingFactory(): Unit = {
val props = properties()
+ props.put(CONNECTOR_TYPE, "fixed")
+ props.put(FORMAT_TYPE, "test")
props.put("failing", "true")
TableSourceFactoryService.findAndCreateTableSource(props.toMap)
}
+ @Test
+ def testWildcardFormat(): Unit = {
+ val props = properties()
+ props.put(CONNECTOR_TYPE, "wildcard")
+ props.put(FORMAT_TYPE, "test")
+ props.put("format.type", "not-test")
+ props.put("format.not-test-property", "wildcard-property")
+ TableSourceFactoryService.findAndCreateTableSource(props.toMap)
--- End diff --
assert what was found?
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200656012
--- Diff: flink-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceFactoryTestBase.java ---
@@ -0,0 +1,175 @@
+/*
+ * 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.streaming.connectors.kafka;
+
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.streaming.connectors.kafka.config.StartupMode;
+import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.api.Types;
+import org.apache.flink.table.descriptors.Kafka;
+import org.apache.flink.table.descriptors.Rowtime;
+import org.apache.flink.table.descriptors.Schema;
+import org.apache.flink.table.descriptors.TestTableSourceDescriptor;
+import org.apache.flink.table.formats.utils.TestDeserializationSchema;
+import org.apache.flink.table.formats.utils.TestTableFormat;
+import org.apache.flink.table.sources.RowtimeAttributeDescriptor;
+import org.apache.flink.table.sources.TableSource;
+import org.apache.flink.table.sources.TableSourceFactoryService;
+import org.apache.flink.table.sources.tsextractors.ExistingField;
+import org.apache.flink.table.sources.wmstrategies.AscendingTimestamps;
+import org.apache.flink.types.Row;
+
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Abstract test base for {@link KafkaTableSourceFactory}.
+ */
+public abstract class KafkaTableSourceFactoryTestBase {
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testTableSource() {
+
+ // prepare parameters for Kafka table source
+
+ final TableSchema schema = TableSchema.builder()
+ .field("fruit-name", Types.STRING())
+ .field("count", Types.DECIMAL())
+ .field("event-time", Types.SQL_TIMESTAMP())
+ .field("proc-time", Types.SQL_TIMESTAMP())
+ .build();
+
+ final String proctimeAttribute = "proc-time";
+
+ final List<RowtimeAttributeDescriptor> rowtimeAttributeDescriptors = Collections.singletonList(
+ new RowtimeAttributeDescriptor("event-time", new ExistingField("time"), new AscendingTimestamps()));
+
+ final Map<String, String> fieldMapping = new HashMap<>();
+ fieldMapping.put("fruit-name", "name");
+ fieldMapping.put("count", "count");
+
+ final String topic = "myTopic";
+
+ final Properties kafkaProperties = new Properties();
+ kafkaProperties.setProperty("zookeeper.connect", "dummy");
+ kafkaProperties.setProperty("group.id", "dummy");
+
+ final Map<KafkaTopicPartition, Long> specificOffsets = new HashMap<>();
+ specificOffsets.put(new KafkaTopicPartition(topic, 0), 100L);
+ specificOffsets.put(new KafkaTopicPartition(topic, 1), 123L);
+
+ final TestDeserializationSchema deserializationSchema = new TestDeserializationSchema(
+ TableSchema.builder()
+ .field("name", Types.STRING())
+ .field("count", Types.DECIMAL())
+ .field("time", Types.SQL_TIMESTAMP())
+ .build()
+ .toRowType()
+ );
+
+ final StartupMode startupMode = StartupMode.SPECIFIC_OFFSETS;
+
+ final KafkaTableSource expected = getKafkaTableSource(
+ schema,
+ proctimeAttribute,
+ rowtimeAttributeDescriptors,
+ fieldMapping,
+ topic,
+ kafkaProperties,
+ deserializationSchema,
+ startupMode,
+ specificOffsets);
+
+ // construct table source using descriptors and table source factory
+
+ final Map<Integer, Long> offsets = new HashMap<>();
+ offsets.put(0, 100L);
+ offsets.put(1, 123L);
+
+ final TestTableSourceDescriptor testDesc = new TestTableSourceDescriptor(
+ new Kafka()
+ .version(getKafkaVersion())
+ .topic(topic)
+ .properties(kafkaProperties)
+ .startFromSpecificOffsets(offsets))
+ .addFormat(new TestTableFormat())
+ .addSchema(
+ new Schema()
+ .field("fruit-name", Types.STRING()).from("name")
+ .field("count", Types.DECIMAL()) // no from so it must match with the input
+ .field("event-time", Types.SQL_TIMESTAMP()).rowtime(
+ new Rowtime().timestampsFromField("time").watermarksPeriodicAscending())
+ .field("proc-time", Types.SQL_TIMESTAMP()).proctime());
+
+ final TableSource<?> foundSource = TableSourceFactoryService.findAndCreateTableSource(testDesc);
+
+ assertEquals(expected, foundSource);
+
+ // test Kafka consumer
+ final KafkaTableSource kafkaSource = (KafkaTableSource) foundSource;
--- End diff --
`actualKafkaSource` or inline this cast with line above
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200686436
--- Diff: flink-connectors/flink-connector-kafka-base/src/main/java/org/apache/flink/streaming/connectors/kafka/KafkaTableSourceFactory.java ---
@@ -124,110 +123,130 @@
params.putProperties(properties);
// validate
- new SchemaValidator(true).validate(params);
+ // allow Kafka timestamps to be used, watermarks can not be received from source
+ new SchemaValidator(true, supportsKafkaTimestamps(), false).validate(params);
new KafkaValidator().validate(params);
- formatValidator().validate(params);
- // build
- final KafkaTableSource.Builder builder = createBuilderWithFormat(params);
+ // deserialization schema using format discovery
+ final DeserializationSchemaFactory<?> formatFactory = TableFormatFactoryService.find(
+ DeserializationSchemaFactory.class,
+ properties,
+ this.getClass().getClassLoader());
+ @SuppressWarnings("unchecked")
+ final DeserializationSchema<Row> deserializationSchema = (DeserializationSchema<Row>) formatFactory
+ .createDeserializationSchema(properties);
- // topic
- final String topic = params.getString(CONNECTOR_TOPIC);
- builder.forTopic(topic);
+ // schema
+ final TableSchema schema = params.getTableSchema(SCHEMA());
+
+ // proctime
+ final String proctimeAttribute = SchemaValidator.deriveProctimeAttribute(params).orElse(null);
+
+ // rowtime
+ final List<RowtimeAttributeDescriptor> rowtimeAttributes = SchemaValidator.deriveRowtimeAttributes(params);
+ if (rowtimeAttributes.size() > 1) {
--- End diff --
Good point. I will remove this check.
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r201028445
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/formats/TableFormatFactory.scala ---
@@ -0,0 +1,85 @@
+/*
+ * 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.formats
+
+import java.util
+
+import org.apache.flink.api.common.serialization.{DeserializationSchema, SerializationSchema}
+
+/**
+ * A factory to create different table format instances. This factory is used with Java's Service
+ * Provider Interfaces (SPI) for discovering. A factory is called with a set of normalized
+ * properties that describe the desired format. The factory allows for matching to the given set of
+ * properties. See also [[SerializationSchemaFactory]] and [[DeserializationSchemaFactory]] for
+ * creating configured instances of format classes accordingly.
+ *
+ * Classes that implement this interface need to be added to the
+ * "META_INF/services/org.apache.flink.table.formats.TableFormatFactory' file of a JAR file in
+ * the current classpath to be found.
+ *
+ * @tparam T record type that the format produces or consumes
+ */
+trait TableFormatFactory[T] {
+
+ /**
+ * Specifies the context that this factory has been implemented for. The framework guarantees
+ * to only use the factory if the specified set of properties and values are met.
+ *
+ * Typical properties might be:
+ * - format.type
+ * - format.version
+ *
+ * Specified property versions allow the framework to provide backwards compatible properties
+ * in case of string format changes:
+ * - format.property-version
+ *
+ * An empty context means that the factory matches for all requests.
+ */
+ def requiredContext(): util.Map[String, String]
--- End diff --
I don't understand `Context` in this context. Rename to `requiredProperties`?
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r201008748
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/formats/TableFormatFactoryService.scala ---
@@ -0,0 +1,245 @@
+/*
+ * 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.formats
+
+import java.util.{ServiceConfigurationError, ServiceLoader, Map => JMap}
+
+import org.apache.flink.table.api._
+import org.apache.flink.table.descriptors.FormatDescriptorValidator.FORMAT_PROPERTY_VERSION
+import org.apache.flink.table.descriptors._
+import org.apache.flink.table.util.Logging
+
+import _root_.scala.collection.JavaConverters._
+import _root_.scala.collection.mutable
+
+/**
+ * Service provider interface for finding a suitable [[TableFormatFactory]] for the
+ * given properties.
+ */
+object TableFormatFactoryService extends Logging {
--- End diff --
Please deduplicate this code with `TableSourceFactoryService`. I haven't looked into this thoroughly so I'm not sure if you should extract some static methods or introduce a base class, but there are some code duplications like:
```
val iter = ...
val requiredContextJava = ...
val plainContext = ...
val plainProperties = ...
```
etc.
---
[GitHub] flink pull request #6264: [FLINK-8558] [table] Add unified format interfaces...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/6264#discussion_r200706630
--- Diff: flink-libraries/flink-table/src/main/scala/org/apache/flink/table/formats/TableFormatFactoryService.scala ---
@@ -0,0 +1,245 @@
+/*
+ * 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.formats
+
+import java.util.{ServiceConfigurationError, ServiceLoader, Map => JMap}
+
+import org.apache.flink.table.api._
+import org.apache.flink.table.descriptors.FormatDescriptorValidator.FORMAT_PROPERTY_VERSION
+import org.apache.flink.table.descriptors._
+import org.apache.flink.table.util.Logging
+
+import _root_.scala.collection.JavaConverters._
+import _root_.scala.collection.mutable
+
+/**
+ * Service provider interface for finding a suitable [[TableFormatFactory]] for the
+ * given properties.
+ */
+object TableFormatFactoryService extends Logging {
+
+ private lazy val defaultLoader = ServiceLoader.load(classOf[TableFormatFactory[_]])
+
+ /**
+ * Finds a table format factory of the given class and creates configured instances from the
+ * given descriptor.
+ *
+ * @param factoryClass desired format factory
+ * @param descriptor descriptor that describes the format
+ * @tparam T factory class type
+ * @return configured instance from factory
+ */
+ def find[T](factoryClass: Class[T], descriptor: Descriptor): T = {
+ find(factoryClass, descriptor, null)
+ }
+
+ /**
+ * Finds a table format factory of the given class and creates configured instances from the
+ * given descriptor and classloader.
+ *
+ * @param factoryClass desired format factory
+ * @param descriptor descriptor that describes the format
+ * @param classLoader classloader for service loading
+ * @tparam T factory class type
+ * @return configured instance from factory
+ */
+ def find[T](
+ factoryClass: Class[T],
+ descriptor: Descriptor,
+ classLoader: ClassLoader)
+ : T = {
+
+ val properties = new DescriptorProperties()
+ descriptor.addProperties(properties)
+ find(factoryClass, properties.asMap, classLoader)
+ }
+
+ /**
+ * Finds a table format factory of the given class and creates configured instances from the
+ * given property map.
+ *
+ * @param factoryClass desired format factory
+ * @param propertyMap properties that describes the format
+ * @tparam T factory class type
+ * @return configured instance from factory
+ */
+ def find[T](factoryClass: Class[T], propertyMap: JMap[String, String]): T = {
+ find(factoryClass, propertyMap, null)
+ }
+
+ /**
+ * Finds a table format factory of the given class and creates configured instances from the
+ * given property map and classloader.
+ *
+ * @param factoryClass desired format factory
+ * @param propertyMap properties that describes the format
+ * @param classLoader classloader for service loading
+ * @tparam T factory class type
+ * @return configured instance from factory
+ */
+ def find[T](
+ factoryClass: Class[T],
+ propertyMap: JMap[String, String],
+ classLoader: ClassLoader)
+ : T = {
+
+ val properties = propertyMap.asScala.toMap
+
+ // find matching context
+ val (foundFactories, contextFactories) = findMatchingContext(properties, classLoader)
+ if (contextFactories.isEmpty) {
+ throw new NoMatchingTableFormatException(
+ "No context matches.",
+ factoryClass,
+ foundFactories,
+ properties)
+ }
+
+ // filter for factory class
+ val classFactories = contextFactories.filter(f => factoryClass.isAssignableFrom(f.getClass))
+ if (classFactories.isEmpty) {
+ throw new NoMatchingTableFormatException(
+ s"No factory implements '${factoryClass.getCanonicalName}'.",
+ factoryClass,
+ foundFactories,
+ properties)
+ }
+
+ // filter by supported keys
+ val plainGivenKeys = mutable.ArrayBuffer[String]()
+ properties.keys.foreach { k =>
+ // replace arrays with wildcard
+ val key = k.replaceAll(".\\d+", ".#")
+ // ignore duplicates
+ if (!plainGivenKeys.contains(key)) {
+ plainGivenKeys += key
+ }
+ }
+ var lastKey: Option[String] = None
+ val supportedFactories = classFactories.filter { factory =>
+ val requiredContextKeys = normalizeContext(factory).keySet
+ val includeSchema = factory.supportsSchemaDerivation()
+ val supportedKeys = normalizeSupportedProperties(factory)
+ val givenKeys = plainGivenKeys
+ // ignore context keys
+ .filter(!requiredContextKeys.contains(_))
+ // ignore non-format (or schema) keys
+ .filter { k =>
+ if (includeSchema) {
+ k.startsWith(SchemaValidator.SCHEMA + ".") ||
+ k.startsWith(FormatDescriptorValidator.FORMAT + ".")
+ } else {
+ k.startsWith(FormatDescriptorValidator.FORMAT + ".")
+ }
+ }
+ givenKeys.forall { k =>
+ lastKey = Option(k)
+ supportedKeys.contains(k)
+ }
+ }
+ if (supportedFactories.isEmpty && classFactories.length == 1 && lastKey.isDefined) {
+ // special case: when there is only one matching factory but the last property key
+ // was incorrect
+ val factory = classFactories.head
+ val supportedKeys = normalizeSupportedProperties(factory)
+ throw new NoMatchingTableFormatException(
+ s"""
+ |The matching factory '${factory.getClass.getName}' doesn't support '${lastKey.get}'.
+ |
+ |Supported properties of this factory are:
+ |${supportedKeys.sorted.mkString("\n")}""".stripMargin,
+ factoryClass,
+ foundFactories,
+ properties)
+ } else if (supportedFactories.isEmpty) {
+ throw new NoMatchingTableFormatException(
+ s"No factory supports all properties.",
+ factoryClass,
+ foundFactories,
+ properties)
+ } else if (supportedFactories.length > 1) {
+ throw new AmbiguousTableFormatException(
+ supportedFactories,
+ factoryClass,
+ foundFactories,
+ properties)
+ }
+
+ supportedFactories.head.asInstanceOf[T]
+ }
+
+ private def findMatchingContext(
+ properties: Map[String, String],
+ classLoader: ClassLoader)
+ : (Seq[TableFormatFactory[_]], Seq[TableFormatFactory[_]]) = {
+
+ val foundFactories = mutable.ArrayBuffer[TableFormatFactory[_]]()
+ val matchingFactories = mutable.ArrayBuffer[TableFormatFactory[_]]()
+
+ try {
+ val iter = if (classLoader == null) {
+ defaultLoader.iterator()
+ } else {
+ val customLoader = ServiceLoader.load(classOf[TableFormatFactory[_]], classLoader)
+ customLoader.iterator()
+ }
+ while (iter.hasNext) {
+ val factory = iter.next()
+ foundFactories += factory
+
+ val requestedContext = normalizeContext(factory)
+
+ val plainContext = mutable.Map[String, String]()
+ plainContext ++= requestedContext
+ // we remove the version for now until we have the first backwards compatibility case
+ // with the version we can provide mappings in case the format changes
+ plainContext.remove(FORMAT_PROPERTY_VERSION)
+
+ // check if required context is met
+ if (plainContext.forall(e => properties.contains(e._1) && properties(e._1) == e._2)) {
+ matchingFactories += factory
+ }
+ }
+ } catch {
+ case e: ServiceConfigurationError =>
+ LOG.error("Could not load service provider for table format factories.", e)
+ throw new TableException("Could not load service provider for table format factories.", e)
+ }
+
+ (foundFactories, matchingFactories)
+ }
+
+ private def normalizeContext(factory: TableFormatFactory[_]): Map[String, String] = {
+ val requiredContextJava = factory.requiredContext()
+ if (requiredContextJava != null) {
+ requiredContextJava.asScala.map(e => (e._1.toLowerCase, e._2)).toMap
+ } else {
+ Map[String, String]()
+ }
+ }
+
+ private def normalizeSupportedProperties(factory: TableFormatFactory[_]): Seq[String] = {
+ val supportedPropertiesJava = factory.supportedProperties()
+ if (supportedPropertiesJava != null) {
--- End diff --
This is just a safety check. Because factories might be implemented by users. According to the documentation this should not be null. But it does hurt to check for null on API borders.
---