[GitHub] [flink] JingsongLi commented on a change in pull request #18088: [FLINK-25174][table] Introduce managed table interfaces and callback
JingsongLi commented on a change in pull request #18088:
URL: https://github.com/apache/flink/pull/18088#discussion_r77853
##
File path:
flink-table/flink-table-common/src/main/java/org/apache/flink/table/factories/FactoryUtil.java
##
@@ -620,10 +620,16 @@ public static String getFormatPrefix(
Class factoryClass, DynamicTableFactory.Context context) {
final String connectorOption =
context.getCatalogTable().getOptions().get(CONNECTOR.key());
if (connectorOption == null) {
-throw new ValidationException(
-String.format(
-"Table options do not contain an option key '%s'
for discovering a connector.",
-CONNECTOR.key()));
+ManagedTableFactory factory =
Review comment:
You have convinced me that I think we can use `default`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [flink] JingsongLi commented on a change in pull request #18088: [FLINK-25174][table] Introduce managed table interfaces and callback
JingsongLi commented on a change in pull request #18088:
URL: https://github.com/apache/flink/pull/18088#discussion_r770192030
##
File path:
flink-table/flink-table-common/src/main/java/org/apache/flink/table/factories/FactoryUtil.java
##
@@ -620,10 +624,16 @@ public static String getFormatPrefix(
Class factoryClass, DynamicTableFactory.Context context) {
final String connectorOption =
context.getCatalogTable().getOptions().get(CONNECTOR.key());
if (connectorOption == null) {
-throw new ValidationException(
-String.format(
-"Table options do not contain an option key '%s'
for discovering a connector.",
-CONNECTOR.key()));
+ManagedTableFactory factory =
+
ManagedTableFactory.discoverManagedTableFactory(context.getClassLoader());
+if (!factoryClass.isAssignableFrom(factory.getClass())) {
+throw new ValidationException(
+String.format(
+"The managed table factory '%s' dose not
implement '%s'.",
Review comment:
I created `FactoryUtil.discoverManagedTableFactory`, it throws exception
messages for `not found` and `multiple found`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [flink] JingsongLi commented on a change in pull request #18088: [FLINK-25174][table] Introduce managed table interfaces and callback
JingsongLi commented on a change in pull request #18088:
URL: https://github.com/apache/flink/pull/18088#discussion_r770182123
##
File path:
flink-table/flink-table-common/src/main/java/org/apache/flink/table/factories/FactoryUtil.java
##
@@ -620,10 +624,16 @@ public static String getFormatPrefix(
Class factoryClass, DynamicTableFactory.Context context) {
final String connectorOption =
context.getCatalogTable().getOptions().get(CONNECTOR.key());
if (connectorOption == null) {
-throw new ValidationException(
-String.format(
-"Table options do not contain an option key '%s'
for discovering a connector.",
-CONNECTOR.key()));
+ManagedTableFactory factory =
+
ManagedTableFactory.discoverManagedTableFactory(context.getClassLoader());
+if (!factoryClass.isAssignableFrom(factory.getClass())) {
+throw new ValidationException(
+String.format(
+"The managed table factory '%s' dose not
implement '%s'.",
Review comment:
Nice! I prefer to throw new exception directly, we don't need repetitive
old exception.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [flink] JingsongLi commented on a change in pull request #18088: [FLINK-25174][table] Introduce managed table interfaces and callback
JingsongLi commented on a change in pull request #18088:
URL: https://github.com/apache/flink/pull/18088#discussion_r770178624
##
File path:
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogManager.java
##
@@ -686,7 +701,14 @@ public void createTemporaryTable(
}
return v;
} else {
-final CatalogBaseTable resolvedTable =
resolveCatalogBaseTable(table);
+ResolvedCatalogBaseTable resolvedTable =
+managedTableListener.notifyTableCreation(
+
getCatalog(objectIdentifier.getCatalogName()).orElse(null),
+objectIdentifier,
+resolveCatalogBaseTable(table),
+true,
Review comment:
I got it. Here is for temporary managed table.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [flink] JingsongLi commented on a change in pull request #18088: [FLINK-25174][table] Introduce managed table interfaces and callback
JingsongLi commented on a change in pull request #18088:
URL: https://github.com/apache/flink/pull/18088#discussion_r769442993
##
File path:
flink-table/flink-table-common/src/main/java/org/apache/flink/table/factories/FactoryUtil.java
##
@@ -620,10 +620,16 @@ public static String getFormatPrefix(
Class factoryClass, DynamicTableFactory.Context context) {
final String connectorOption =
context.getCatalogTable().getOptions().get(CONNECTOR.key());
if (connectorOption == null) {
-throw new ValidationException(
-String.format(
-"Table options do not contain an option key '%s'
for discovering a connector.",
-CONNECTOR.key()));
+ManagedTableFactory factory =
Review comment:
I prefer the empty string to other ids because the empty string itself
is a special symbol that is often mixed up with "none".
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [flink] JingsongLi commented on a change in pull request #18088: [FLINK-25174][table] Introduce managed table interfaces and callback
JingsongLi commented on a change in pull request #18088:
URL: https://github.com/apache/flink/pull/18088#discussion_r769441143
##
File path:
flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/Catalog.java
##
@@ -311,6 +311,15 @@ void createTable(ObjectPath tablePath, CatalogBaseTable
table, boolean ignoreIfE
void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean
ignoreIfNotExists)
throws TableNotExistException, CatalogException;
+/**
+ * If true, tables which do not specify a connector will be translated to
managed tables.
+ *
+ * @see CatalogBaseTable.TableKind#MANAGED
+ */
+default boolean supportsManagedTable() {
Review comment:
You can see the first few replies of the FLIP discussion.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [flink] JingsongLi commented on a change in pull request #18088: [FLINK-25174][table] Introduce managed table interfaces and callback
JingsongLi commented on a change in pull request #18088:
URL: https://github.com/apache/flink/pull/18088#discussion_r769432588
##
File path:
flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/Catalog.java
##
@@ -311,6 +311,15 @@ void createTable(ObjectPath tablePath, CatalogBaseTable
table, boolean ignoreIfE
void alterTable(ObjectPath tablePath, CatalogBaseTable newTable, boolean
ignoreIfNotExists)
throws TableNotExistException, CatalogException;
+/**
+ * If true, tables which do not specify a connector will be translated to
managed tables.
+ *
+ * @see CatalogBaseTable.TableKind#MANAGED
+ */
+default boolean supportsManagedTable() {
Review comment:
There are some catalogs that already use table without connector as
their catalog internal table.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [flink] JingsongLi commented on a change in pull request #18088: [FLINK-25174][table] Introduce managed table interfaces and callback
JingsongLi commented on a change in pull request #18088:
URL: https://github.com/apache/flink/pull/18088#discussion_r769195266
##
File path:
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogManager.java
##
@@ -686,7 +701,14 @@ public void createTemporaryTable(
}
return v;
} else {
-final CatalogBaseTable resolvedTable =
resolveCatalogBaseTable(table);
+ResolvedCatalogBaseTable resolvedTable =
+managedTableListener.notifyTableCreation(
+
getCatalog(objectIdentifier.getCatalogName()).orElse(null),
+objectIdentifier,
+resolveCatalogBaseTable(table),
+true,
Review comment:
Our temporary tables can be catalog-owned, I remember this is why we do:
https://issues.apache.org/jira/browse/FLINK-21185
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [flink] JingsongLi commented on a change in pull request #18088: [FLINK-25174][table] Introduce managed table interfaces and callback
JingsongLi commented on a change in pull request #18088:
URL: https://github.com/apache/flink/pull/18088#discussion_r768295184
##
File path:
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/ManagedTableListener.java
##
@@ -0,0 +1,125 @@
+/*
+ * 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.catalog;
+
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.factories.DynamicTableFactory;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.util.StringUtils;
+
+import javax.annotation.Nullable;
+
+import java.util.Map;
+
+import static
org.apache.flink.table.catalog.CatalogBaseTable.TableKind.MANAGED;
+import static
org.apache.flink.table.factories.ManagedTableFactory.discoverManagedTableFactory;
+
+/** The listener for managed table operations. */
+public class ManagedTableListener {
+
+private final ClassLoader classLoader;
+
+private final ReadableConfig config;
+
+public ManagedTableListener(ClassLoader classLoader, ReadableConfig
config) {
+this.classLoader = classLoader;
+this.config = config;
+}
+
+/** Notify for creating managed table. */
+public ResolvedCatalogBaseTable notifyTableCreation(
+@Nullable Catalog catalog,
+ObjectIdentifier identifier,
+ResolvedCatalogBaseTable table,
+boolean isTemporary,
+boolean ignoreIfExists) {
+if (isNewlyManagedTable(catalog, table)) {
+ResolvedCatalogTable managedTable = createManagedTable(identifier,
table, isTemporary);
+discoverManagedTableFactory(classLoader)
+.onCreateTable(
+createTableFactoryContext(identifier,
managedTable, isTemporary),
+ignoreIfExists);
+return managedTable;
+}
+return table;
+}
+
+/** Notify for dropping managed table. */
+public void notifyTableDrop(
+ObjectIdentifier identifier,
+ResolvedCatalogBaseTable table,
+boolean isTemporary,
+boolean ignoreIfNotExists) {
+if (table.getTableKind() == MANAGED) {
+discoverManagedTableFactory(classLoader)
+.onDropTable(
+createTableFactoryContext(
+identifier, (ResolvedCatalogTable) table,
isTemporary),
+ignoreIfNotExists);
+}
+}
+
+private boolean isNewlyManagedTable(
+@Nullable Catalog catalog, ResolvedCatalogBaseTable table) {
+if (catalog == null || !catalog.supportsManagedTable()) {
+// catalog not support managed table
+return false;
+}
+if (table.getTableKind() == CatalogBaseTable.TableKind.VIEW) {
+// view is not managed table
+return false;
+}
+if (!StringUtils.isNullOrWhitespaceOnly(
+table.getOptions().get(FactoryUtil.CONNECTOR.key( {
+// with connector is not managed table
+return false;
+}
+CatalogBaseTable origin = table.getOrigin();
+// only DefaultCatalogTable or CatalogTableImpl is managed table
+return origin instanceof DefaultCatalogTable || origin instanceof
CatalogTableImpl;
Review comment:
For `InlineCatalogTable`, it is a little tricky like:
```
try {
origin.getOptions();
} catch (TableException ignore) {
// exclude abnormal tables, such as InlineCatalogTable that does not have
the options
return false;
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [flink] JingsongLi commented on a change in pull request #18088: [FLINK-25174][table] Introduce managed table interfaces and callback
JingsongLi commented on a change in pull request #18088:
URL: https://github.com/apache/flink/pull/18088#discussion_r768295184
##
File path:
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/ManagedTableListener.java
##
@@ -0,0 +1,125 @@
+/*
+ * 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.catalog;
+
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.factories.DynamicTableFactory;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.util.StringUtils;
+
+import javax.annotation.Nullable;
+
+import java.util.Map;
+
+import static
org.apache.flink.table.catalog.CatalogBaseTable.TableKind.MANAGED;
+import static
org.apache.flink.table.factories.ManagedTableFactory.discoverManagedTableFactory;
+
+/** The listener for managed table operations. */
+public class ManagedTableListener {
+
+private final ClassLoader classLoader;
+
+private final ReadableConfig config;
+
+public ManagedTableListener(ClassLoader classLoader, ReadableConfig
config) {
+this.classLoader = classLoader;
+this.config = config;
+}
+
+/** Notify for creating managed table. */
+public ResolvedCatalogBaseTable notifyTableCreation(
+@Nullable Catalog catalog,
+ObjectIdentifier identifier,
+ResolvedCatalogBaseTable table,
+boolean isTemporary,
+boolean ignoreIfExists) {
+if (isNewlyManagedTable(catalog, table)) {
+ResolvedCatalogTable managedTable = createManagedTable(identifier,
table, isTemporary);
+discoverManagedTableFactory(classLoader)
+.onCreateTable(
+createTableFactoryContext(identifier,
managedTable, isTemporary),
+ignoreIfExists);
+return managedTable;
+}
+return table;
+}
+
+/** Notify for dropping managed table. */
+public void notifyTableDrop(
+ObjectIdentifier identifier,
+ResolvedCatalogBaseTable table,
+boolean isTemporary,
+boolean ignoreIfNotExists) {
+if (table.getTableKind() == MANAGED) {
+discoverManagedTableFactory(classLoader)
+.onDropTable(
+createTableFactoryContext(
+identifier, (ResolvedCatalogTable) table,
isTemporary),
+ignoreIfNotExists);
+}
+}
+
+private boolean isNewlyManagedTable(
+@Nullable Catalog catalog, ResolvedCatalogBaseTable table) {
+if (catalog == null || !catalog.supportsManagedTable()) {
+// catalog not support managed table
+return false;
+}
+if (table.getTableKind() == CatalogBaseTable.TableKind.VIEW) {
+// view is not managed table
+return false;
+}
+if (!StringUtils.isNullOrWhitespaceOnly(
+table.getOptions().get(FactoryUtil.CONNECTOR.key( {
+// with connector is not managed table
+return false;
+}
+CatalogBaseTable origin = table.getOrigin();
+// only DefaultCatalogTable or CatalogTableImpl is managed table
+return origin instanceof DefaultCatalogTable || origin instanceof
CatalogTableImpl;
Review comment:
For `InlineCatalogTable`, it is a little tricky like:
```
try {
origin.getOptions();
} catch (TableException ignore) {
// exclude abnormal tables, such as InlineCatalogTable that does
not have the options
return false;
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [flink] JingsongLi commented on a change in pull request #18088: [FLINK-25174][table] Introduce managed table interfaces and callback
JingsongLi commented on a change in pull request #18088:
URL: https://github.com/apache/flink/pull/18088#discussion_r767571857
##
File path:
flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/DefaultCatalogTable.java
##
@@ -39,16 +39,19 @@
private final @Nullable String comment;
private final List partitionKeys;
private final Map options;
+private final TableKind tableKind;
protected DefaultCatalogTable(
Schema schema,
@Nullable String comment,
List partitionKeys,
-Map options) {
+Map options,
+TableKind tableKind) {
this.schema = checkNotNull(schema, "Schema must not be null.");
this.comment = comment;
this.partitionKeys = checkNotNull(partitionKeys, "Partition keys must
not be null.");
this.options = checkNotNull(options, "Options must not be null.");
+this.tableKind = checkNotNull(tableKind, "Table kind must not be
null.");
Review comment:
We can introduce a `DefaultManagedTable`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [flink] JingsongLi commented on a change in pull request #18088: [FLINK-25174][table] Introduce managed table interfaces and callback
JingsongLi commented on a change in pull request #18088:
URL: https://github.com/apache/flink/pull/18088#discussion_r767569684
##
File path:
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/ManagedTableListener.java
##
@@ -0,0 +1,125 @@
+/*
+ * 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.catalog;
+
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.factories.DynamicTableFactory;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.util.StringUtils;
+
+import javax.annotation.Nullable;
+
+import java.util.Map;
+
+import static
org.apache.flink.table.catalog.CatalogBaseTable.TableKind.MANAGED;
+import static
org.apache.flink.table.factories.ManagedTableFactory.discoverManagedTableFactory;
+
+/** The listener for managed table operations. */
+public class ManagedTableListener {
+
+private final ClassLoader classLoader;
+
+private final ReadableConfig config;
+
+public ManagedTableListener(ClassLoader classLoader, ReadableConfig
config) {
+this.classLoader = classLoader;
+this.config = config;
+}
+
+/** Notify for creating managed table. */
+public ResolvedCatalogBaseTable notifyTableCreation(
+@Nullable Catalog catalog,
+ObjectIdentifier identifier,
+ResolvedCatalogBaseTable table,
+boolean isTemporary,
+boolean ignoreIfExists) {
+if (isNewlyManagedTable(catalog, table)) {
+ResolvedCatalogTable managedTable = createManagedTable(identifier,
table, isTemporary);
+discoverManagedTableFactory(classLoader)
+.onCreateTable(
+createTableFactoryContext(identifier,
managedTable, isTemporary),
+ignoreIfExists);
+return managedTable;
+}
+return table;
+}
+
+/** Notify for dropping managed table. */
+public void notifyTableDrop(
+ObjectIdentifier identifier,
+ResolvedCatalogBaseTable table,
+boolean isTemporary,
+boolean ignoreIfNotExists) {
+if (table.getTableKind() == MANAGED) {
+discoverManagedTableFactory(classLoader)
+.onDropTable(
+createTableFactoryContext(
+identifier, (ResolvedCatalogTable) table,
isTemporary),
+ignoreIfNotExists);
+}
+}
+
+private boolean isNewlyManagedTable(
+@Nullable Catalog catalog, ResolvedCatalogBaseTable table) {
+if (catalog == null || !catalog.supportsManagedTable()) {
+// catalog not support managed table
+return false;
+}
+if (table.getTableKind() == CatalogBaseTable.TableKind.VIEW) {
+// view is not managed table
+return false;
+}
+if (!StringUtils.isNullOrWhitespaceOnly(
+table.getOptions().get(FactoryUtil.CONNECTOR.key( {
+// with connector is not managed table
+return false;
+}
+CatalogBaseTable origin = table.getOrigin();
+// only DefaultCatalogTable or CatalogTableImpl is managed table
+return origin instanceof DefaultCatalogTable || origin instanceof
CatalogTableImpl;
Review comment:
You are right, we should exclude `InlineCatalogTable` and
`ConnectorCatalogTable` these two troublemakers.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
