You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by sn...@apache.org on 2025/01/07 21:47:12 UTC
(flink) branch master updated (ae7184bd222 -> 503ad9b7b8e)
This is an automated email from the ASF dual-hosted git repository.
snuyanzin pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
from ae7184bd222 [FLINK-36070][runtime/web] Show pending operators in overview page.
new d432dff3a9e [FLINK-37007][table] Add missing createView methods to TableEnvironment
new 503ad9b7b8e [hotfix][docs] Add python docs for `create_table`/`create_temporary_table`
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../docs/dev/python/table/table_environment.md | 33 +++++++++++++
.../docs/dev/python/table/table_environment.md | 33 +++++++++++++
.../reference/pyflink.table/table_environment.rst | 1 +
flink-python/pyflink/table/table_environment.py | 23 +++++++++
.../apache/flink/table/api/TableEnvironment.java | 34 +++++++++++++-
.../table/api/internal/TableEnvironmentImpl.java | 23 ++++++++-
.../apache/flink/table/catalog/CatalogManager.java | 12 ++++-
.../flink/table/api/TableEnvironmentTest.java | 54 +++++++++++++++++++++-
.../planner/runtime/stream/table/ValuesITCase.java | 17 +++++--
9 files changed, 220 insertions(+), 10 deletions(-)
(flink) 02/02: [hotfix][docs] Add python docs for `create_table`/`create_temporary_table`
Posted by sn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
snuyanzin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 503ad9b7b8e969b17fce6569efa0e7566938a941
Author: Sergey Nuyanzin <sn...@gmail.com>
AuthorDate: Tue Jan 7 00:06:05 2025 +0100
[hotfix][docs] Add python docs for `create_table`/`create_temporary_table`
---
.../docs/dev/python/table/table_environment.md | 22 ++++++++++++++++++++++
.../docs/dev/python/table/table_environment.md | 22 ++++++++++++++++++++++
2 files changed, 44 insertions(+)
diff --git a/docs/content.zh/docs/dev/python/table/table_environment.md b/docs/content.zh/docs/dev/python/table/table_environment.md
index 527894fdf3e..7016dc76bda 100644
--- a/docs/content.zh/docs/dev/python/table/table_environment.md
+++ b/docs/content.zh/docs/dev/python/table/table_environment.md
@@ -154,6 +154,28 @@ TableEnvironment API
{{< pythondoc file="pyflink.table.html#pyflink.table.TableEnvironment.drop_view" name="link">}}
</td>
</tr>
+ <tr>
+ <td>
+ <strong>create_temporary_table(path, table_descriptor, ignore_if_exists=False)</strong>
+ </td>
+ <td>
+ Registers a `Table` object as a temporary catalog table similar to SQL temporary tables.
+ </td>
+ <td class="text-center">
+ {{< pythondoc file="pyflink.table.html#pyflink.table.TableEnvironment.create_temporary_table" name="link">}}
+ </td>
+ </tr>
+ <tr>
+ <td>
+ <strong>create_table(path, table_descriptor, ignore_if_exists=False)</strong>
+ </td>
+ <td>
+ Registers a `Table` object as a catalog table similar to SQL tables.
+ </td>
+ <td class="text-center">
+ {{< pythondoc file="pyflink.table.html#pyflink.table.TableEnvironment.create_table" name="link">}}
+ </td>
+ </tr>
<tr>
<td>
<strong>drop_temporary_table(table_path)</strong>
diff --git a/docs/content/docs/dev/python/table/table_environment.md b/docs/content/docs/dev/python/table/table_environment.md
index 7ad82b501e5..7441a686495 100644
--- a/docs/content/docs/dev/python/table/table_environment.md
+++ b/docs/content/docs/dev/python/table/table_environment.md
@@ -154,6 +154,28 @@ These APIs are used to create/remove Table API/SQL Tables and write queries:
{{< pythondoc file="pyflink.table.html#pyflink.table.TableEnvironment.drop_view" name="link">}}
</td>
</tr>
+ <tr>
+ <td>
+ <strong>create_temporary_table(path, table_descriptor, ignore_if_exists=False)</strong>
+ </td>
+ <td>
+ Registers a `Table` object as a temporary catalog table similar to SQL temporary tables.
+ </td>
+ <td class="text-center">
+ {{< pythondoc file="pyflink.table.html#pyflink.table.TableEnvironment.create_temporary_table" name="link">}}
+ </td>
+ </tr>
+ <tr>
+ <td>
+ <strong>create_table(path, table_descriptor, ignore_if_exists=False)</strong>
+ </td>
+ <td>
+ Registers a `Table` object as a catalog table similar to SQL tables.
+ </td>
+ <td class="text-center">
+ {{< pythondoc file="pyflink.table.html#pyflink.table.TableEnvironment.create_table" name="link">}}
+ </td>
+ </tr>
<tr>
<td>
<strong>drop_temporary_table(table_path)</strong>
(flink) 01/02: [FLINK-37007][table] Add missing createView methods to TableEnvironment
Posted by sn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
snuyanzin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit d432dff3a9e2c55ecf6c0af898219dccd1e45bfd
Author: Sergey Nuyanzin <sn...@gmail.com>
AuthorDate: Sat Dec 28 09:07:42 2024 +0100
[FLINK-37007][table] Add missing createView methods to TableEnvironment
This closes #25907
---
.../docs/dev/python/table/table_environment.md | 11 +++++
.../docs/dev/python/table/table_environment.md | 11 +++++
.../reference/pyflink.table/table_environment.rst | 1 +
flink-python/pyflink/table/table_environment.py | 23 +++++++++
.../apache/flink/table/api/TableEnvironment.java | 34 +++++++++++++-
.../table/api/internal/TableEnvironmentImpl.java | 23 ++++++++-
.../apache/flink/table/catalog/CatalogManager.java | 12 ++++-
.../flink/table/api/TableEnvironmentTest.java | 54 +++++++++++++++++++++-
.../planner/runtime/stream/table/ValuesITCase.java | 17 +++++--
9 files changed, 176 insertions(+), 10 deletions(-)
diff --git a/docs/content.zh/docs/dev/python/table/table_environment.md b/docs/content.zh/docs/dev/python/table/table_environment.md
index b11cb632e9b..527894fdf3e 100644
--- a/docs/content.zh/docs/dev/python/table/table_environment.md
+++ b/docs/content.zh/docs/dev/python/table/table_environment.md
@@ -121,6 +121,17 @@ TableEnvironment API
{{< pythondoc file="pyflink.table.html#pyflink.table.TableEnvironment.create_temporary_view" name="链接">}}
</td>
</tr>
+ <tr>
+ <td>
+ <strong>create_view(view_path, table, ignore_if_exists=False)</strong>
+ </td>
+ <td>
+ Registers a `Table` object as a view similar to SQL views.
+ </td>
+ <td class="text-center">
+ {{< pythondoc file="pyflink.table.html#pyflink.table.TableEnvironment.create_view" name="link">}}
+ </td>
+ </tr>
<tr>
<td>
<strong>drop_temporary_view(view_path)</strong>
diff --git a/docs/content/docs/dev/python/table/table_environment.md b/docs/content/docs/dev/python/table/table_environment.md
index 5f9aaa7f26a..7ad82b501e5 100644
--- a/docs/content/docs/dev/python/table/table_environment.md
+++ b/docs/content/docs/dev/python/table/table_environment.md
@@ -121,6 +121,17 @@ These APIs are used to create/remove Table API/SQL Tables and write queries:
{{< pythondoc file="pyflink.table.html#pyflink.table.TableEnvironment.create_temporary_view" name="link">}}
</td>
</tr>
+ <tr>
+ <td>
+ <strong>create_view(view_path, table, ignore_if_exists=False)</strong>
+ </td>
+ <td>
+ Registers a `Table` object as a view similar to SQL views.
+ </td>
+ <td class="text-center">
+ {{< pythondoc file="pyflink.table.html#pyflink.table.TableEnvironment.create_view" name="link">}}
+ </td>
+ </tr>
<tr>
<td>
<strong>drop_temporary_view(view_path)</strong>
diff --git a/flink-python/docs/reference/pyflink.table/table_environment.rst b/flink-python/docs/reference/pyflink.table/table_environment.rst
index 95acc613275..46a7060be25 100644
--- a/flink-python/docs/reference/pyflink.table/table_environment.rst
+++ b/flink-python/docs/reference/pyflink.table/table_environment.rst
@@ -161,6 +161,7 @@ keyword, thus must be escaped) in a catalog named 'cat.1' and database named 'db
TableEnvironment.create_temporary_system_function
TableEnvironment.create_temporary_table
TableEnvironment.create_temporary_view
+ TableEnvironment.create_view
TableEnvironment.drop_function
TableEnvironment.drop_table
TableEnvironment.drop_temporary_function
diff --git a/flink-python/pyflink/table/table_environment.py b/flink-python/pyflink/table/table_environment.py
index 8defd3a1f18..7f6784f59b5 100644
--- a/flink-python/pyflink/table/table_environment.py
+++ b/flink-python/pyflink/table/table_environment.py
@@ -1035,6 +1035,29 @@ class TableEnvironment(object):
raise ValueError("Invalid arguments for 'fields': %r" %
','.join([repr(item) for item in fields_or_schema]))
+ def create_view(self,
+ view_path: str,
+ table: Table,
+ ignore_if_exists: Optional[bool] = False):
+ """
+ Registers a :class:`~pyflink.table.Table` API object as a view similar to SQL views.
+
+ This method can only create permanent objects. Temporary objects can shadow permanent ones.
+ If a temporary object exists in a given path,
+ make sure to drop the temporary object first using :func:`drop_temporary_view`.
+
+ :param view_path: The path under which the view will be registered.
+ See also the :class:`~pyflink.table.TableEnvironment` class description
+ for the format of the path.
+ :param table: The view to register.
+ :param ignore_if_exists: If a view or a table exists and the given flag is set,
+ no operation is executed. An exception is thrown otherwise.
+
+ .. versionadded:: 2.0.0
+ """
+
+ self._j_tenv.createView(view_path, table, ignore_if_exists)
+
def add_python_file(self, file_path: str):
"""
Adds a python dependency which could be python files, python packages or
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TableEnvironment.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TableEnvironment.java
index 16bdff3813a..3296c059254 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TableEnvironment.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TableEnvironment.java
@@ -803,8 +803,10 @@ public interface TableEnvironment {
* @param descriptor Template for creating a {@link CatalogTable} instance.
* @param ignoreIfExists If a table exists under the given path and this flag is set, no
* operation is executed. An exception is thrown otherwise.
+ * @return true if table was created in the given path, false if a permanent object already
+ * exists in the given path.
*/
- void createTable(String path, TableDescriptor descriptor, boolean ignoreIfExists);
+ boolean createTable(String path, TableDescriptor descriptor, boolean ignoreIfExists);
/**
* Registers a {@link Table} under a unique name in the TableEnvironment's catalog. Registered
@@ -834,6 +836,36 @@ public interface TableEnvironment {
*/
void createTemporaryView(String path, Table view);
+ /**
+ * Registers a {@link Table} API object as a view similar to SQL views.
+ *
+ * <p>Temporary objects can shadow permanent ones. If a temporary object in a given path exists,
+ * the permanent one will be inaccessible in the current session. To make the permanent object
+ * available again one can drop the corresponding temporary object.
+ *
+ * @param path The path under which the view will be registered. See also the {@link
+ * TableEnvironment} class description for the format of the path.
+ * @param view The view to register.
+ */
+ void createView(String path, Table view);
+
+ /**
+ * Registers a {@link Table} API object as a view similar to SQL views.
+ *
+ * <p>Temporary objects can shadow permanent ones. If a temporary object in a given path exists,
+ * the permanent one will be inaccessible in the current session. To make the permanent object
+ * available again one can drop the corresponding temporary object.
+ *
+ * @param path The path under which the view will be registered. See also the {@link
+ * TableEnvironment} class description for the format of the path.
+ * @param view The view to register.
+ * @param ignoreIfExists If a view or a table exists and the given flag is set, no operation is
+ * executed. An exception is thrown otherwise.
+ * @return true if view was created in the given path, false if a permanent object already
+ * exists in the given path.
+ */
+ boolean createView(String path, Table view, boolean ignoreIfExists);
+
/**
* Scans a registered table and returns the resulting {@link Table}.
*
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
index 5f809f52ac3..3b205333963 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableEnvironmentImpl.java
@@ -506,13 +506,14 @@ public class TableEnvironmentImpl implements TableEnvironmentInternal {
}
@Override
- public void createTable(String path, TableDescriptor descriptor, boolean ignoreIfExists) {
+ public boolean createTable(String path, TableDescriptor descriptor, boolean ignoreIfExists) {
Preconditions.checkNotNull(path, "Path must not be null.");
Preconditions.checkNotNull(descriptor, "Table descriptor must not be null.");
final ObjectIdentifier tableIdentifier =
catalogManager.qualifyIdentifier(getParser().parseIdentifier(path));
- catalogManager.createTable(descriptor.toCatalogTable(), tableIdentifier, ignoreIfExists);
+ return catalogManager.createTable(
+ descriptor.toCatalogTable(), tableIdentifier, ignoreIfExists);
}
@Override
@@ -543,6 +544,24 @@ public class TableEnvironmentImpl implements TableEnvironmentInternal {
catalogManager.createTemporaryTable(tableTable, tableIdentifier, false);
}
+ @Override
+ public void createView(String path, Table view) {
+ createView(path, view, false);
+ }
+
+ @Override
+ public boolean createView(String path, Table view, boolean ignoreIfExists) {
+ Preconditions.checkNotNull(path, "Path must not be null.");
+ Preconditions.checkNotNull(view, "Table view must not be null.");
+ UnresolvedIdentifier identifier = getParser().parseIdentifier(path);
+
+ ObjectIdentifier viewIdentifier = catalogManager.qualifyIdentifier(identifier);
+ QueryOperation queryOperation =
+ qualifyQueryOperation(viewIdentifier, view.getQueryOperation());
+ CatalogBaseTable tableTable = new QueryOperationCatalogView(queryOperation);
+ return catalogManager.createTable(tableTable, viewIdentifier, ignoreIfExists);
+ }
+
@Override
public Table scan(String... tablePath) {
UnresolvedIdentifier unresolvedIdentifier = UnresolvedIdentifier.of(tablePath);
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogManager.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogManager.java
index 029059d3c98..f5efc9e854a 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogManager.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/catalog/CatalogManager.java
@@ -1012,9 +1012,18 @@ public final class CatalogManager implements CatalogRegistry, AutoCloseable {
* @param table The table to put in the given path.
* @param objectIdentifier The fully qualified path where to put the table.
* @param ignoreIfExists If false exception will be thrown if a table exists in the given path.
+ * @return true if table was created in the given path, false if a table already exists in the
+ * given path.
*/
- public void createTable(
+ public boolean createTable(
CatalogBaseTable table, ObjectIdentifier objectIdentifier, boolean ignoreIfExists) {
+ final boolean result;
+ if (ignoreIfExists) {
+ final Optional<CatalogBaseTable> resultOpt = getUnresolvedTable(objectIdentifier);
+ result = resultOpt.isEmpty();
+ } else {
+ result = true;
+ }
execute(
(catalog, path) -> {
ResolvedCatalogBaseTable<?> resolvedTable = resolveCatalogBaseTable(table);
@@ -1045,6 +1054,7 @@ public final class CatalogManager implements CatalogRegistry, AutoCloseable {
objectIdentifier,
false,
"CreateTable");
+ return result;
}
/**
diff --git a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/api/TableEnvironmentTest.java b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/api/TableEnvironmentTest.java
index b449fce3f35..50c53e1f433 100644
--- a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/api/TableEnvironmentTest.java
+++ b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/api/TableEnvironmentTest.java
@@ -20,15 +20,19 @@ package org.apache.flink.table.api;
import org.apache.flink.table.catalog.CatalogBaseTable;
import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.CatalogView;
import org.apache.flink.table.catalog.ContextResolvedTable;
import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
import org.apache.flink.table.operations.SourceQueryOperation;
import org.apache.flink.table.operations.ddl.CreateTableOperation;
import org.apache.flink.table.operations.ddl.DropTableOperation;
import org.apache.flink.table.utils.TableEnvironmentMock;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.util.Map;
import java.util.Optional;
@@ -79,6 +83,54 @@ class TableEnvironmentTest {
assertCreateTableFromDescriptor(tEnv, TEST_SCHEMA, false);
}
+ @ParameterizedTest(name = "{index}: ignoreIfExists ({0})")
+ @ValueSource(booleans = {true, false})
+ void testCreateViewFromTable(final boolean ignoreIfExists) throws Exception {
+ final TableEnvironmentMock tEnv = TableEnvironmentMock.getStreamingInstance();
+
+ final String catalog = tEnv.getCurrentCatalog();
+ final String database = tEnv.getCurrentDatabase();
+
+ tEnv.createTable("T", TEST_DESCRIPTOR);
+
+ assertThat(tEnv.createView("V", tEnv.from("T"), ignoreIfExists)).isTrue();
+
+ final ObjectPath objectPath = new ObjectPath(database, "V");
+
+ final CatalogBaseTable catalogView =
+ tEnv.getCatalog(catalog).orElseThrow(AssertionError::new).getTable(objectPath);
+ assertThat(catalogView).isInstanceOf(CatalogView.class);
+ assertThat(catalogView.getUnresolvedSchema()).isEqualTo(TEST_SCHEMA);
+ }
+
+ @Test
+ void testCreateViewWithSameNameIgnoreIfExists() {
+ final TableEnvironmentMock tEnv = TableEnvironmentMock.getStreamingInstance();
+
+ tEnv.createTable("T", TEST_DESCRIPTOR);
+ tEnv.createView("V", tEnv.from("T"));
+
+ assertThat(tEnv.createView("V", tEnv.from("T"), true)).isFalse();
+ }
+
+ @Test
+ void testCreateViewWithSameName() {
+ final TableEnvironmentMock tEnv = TableEnvironmentMock.getStreamingInstance();
+
+ tEnv.createTable("T", TEST_DESCRIPTOR);
+ tEnv.createView("V", tEnv.from("T"));
+
+ assertThatThrownBy(() -> tEnv.createView("V", tEnv.from("T"), false))
+ .hasCauseInstanceOf(TableAlreadyExistException.class)
+ .hasMessageContaining(
+ "Could not execute CreateTable in path `default_catalog`.`default_database`.`V`");
+
+ assertThatThrownBy(() -> tEnv.createView("V", tEnv.from("T")))
+ .hasCauseInstanceOf(TableAlreadyExistException.class)
+ .hasMessageContaining(
+ "Could not execute CreateTable in path `default_catalog`.`default_database`.`V`");
+ }
+
@Test
void testCreateTableIfNotExistsFromDescriptor() throws Exception {
final TableEnvironmentMock tEnv = TableEnvironmentMock.getStreamingInstance();
@@ -141,7 +193,7 @@ class TableEnvironmentTest {
final String database = tEnv.getCurrentDatabase();
if (ignoreIfExists) {
- tEnv.createTable("T", TEST_DESCRIPTOR, true);
+ assertThat(tEnv.createTable("T", TEST_DESCRIPTOR, true)).isTrue();
} else {
tEnv.createTable("T", TEST_DESCRIPTOR);
}
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/table/ValuesITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/table/ValuesITCase.java
index bcf69fb8220..bf04f175f37 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/table/ValuesITCase.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/stream/table/ValuesITCase.java
@@ -30,6 +30,8 @@ import org.apache.flink.types.Row;
import org.apache.flink.util.CollectionUtil;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.math.BigDecimal;
import java.time.Instant;
@@ -326,18 +328,23 @@ class ValuesITCase extends StreamingTestBase {
assertThat(actual).containsExactlyInAnyOrderElementsOf(expected);
}
- @Test
- void testRegisteringValuesWithComplexTypes() {
+ @ParameterizedTest(name = "{index}: isTemporaryView ({0})")
+ @ValueSource(booleans = {true, false})
+ void testRegisteringValuesWithComplexTypes(final boolean isTemporaryView) {
Map<Integer, Integer> mapData = new HashMap<>();
mapData.put(1, 1);
mapData.put(2, 2);
Row row = Row.of(mapData, Row.of(1, 2, 3), new Integer[] {1, 2});
Table values = tEnv().fromValues(Collections.singletonList(row));
- tEnv().createTemporaryView("values_t", values);
+ final String path = "values_t";
+ if (isTemporaryView) {
+ tEnv().createTemporaryView(path, values);
+ } else {
+ tEnv().createView(path, values);
+ }
List<Row> results =
- CollectionUtil.iteratorToList(
- tEnv().executeSql("select * from values_t").collect());
+ CollectionUtil.iteratorToList(tEnv().executeSql("select * from " + path).collect());
assertThat(results).containsExactly(row);
}