This is an automated email from the ASF dual-hosted git repository.
twalthr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 02510f1116d [FLINK-36809][table] Support ignoreIfExists param for
createTable
02510f1116d is described below
commit 02510f1116d3da371e06c90ff950ca7304e8a5e7
Author: Gustavo de Morais <[email protected]>
AuthorDate: Fri Nov 29 18:03:13 2024 +0100
[FLINK-36809][table] Support ignoreIfExists param for createTable
---
flink-python/pyflink/table/table_environment.py | 22 ++--
.../table/tests/test_table_environment_api.py | 132 ++++++++++++++------
.../apache/flink/table/api/TableEnvironment.java | 61 ++++++++++
.../table/api/internal/TableEnvironmentImpl.java | 16 ++-
.../flink/table/api/TableEnvironmentTest.java | 135 ++++++++++++++-------
5 files changed, 272 insertions(+), 94 deletions(-)
diff --git a/flink-python/pyflink/table/table_environment.py
b/flink-python/pyflink/table/table_environment.py
index 4940b76f75e..3868d487da6 100644
--- a/flink-python/pyflink/table/table_environment.py
+++ b/flink-python/pyflink/table/table_environment.py
@@ -19,7 +19,7 @@ import atexit
import os
import sys
import tempfile
-from typing import Union, List, Tuple, Iterable
+from typing import Union, List, Tuple, Iterable, Optional
from py4j.java_gateway import get_java_class, get_method
@@ -405,7 +405,8 @@ class TableEnvironment(object):
"""
return self._j_tenv.dropTemporaryFunction(path)
- def create_temporary_table(self, path: str, descriptor: TableDescriptor):
+ def create_temporary_table(self, path: str, descriptor: TableDescriptor,
+ ignoreIfExists: Optional[bool] = False):
"""
Registers the given :class:`~pyflink.table.TableDescriptor` as a
temporary catalog table.
@@ -424,16 +425,20 @@ class TableEnvironment(object):
... .build())
... .option("rows-per-second", 10)
... .option("fields.f0.kind", "random")
- ... .build())
+ ... .build(),
+ ... True)
:param path: The path under which the table will be registered.
:param descriptor: Template for creating a 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.
.. versionadded:: 1.14.0
"""
- self._j_tenv.createTemporaryTable(path, descriptor._j_table_descriptor)
+ self._j_tenv.createTemporaryTable(path,
descriptor._j_table_descriptor, ignoreIfExists)
- def create_table(self, path: str, descriptor: TableDescriptor):
+ def create_table(self, path: str, descriptor: TableDescriptor,
+ ignoreIfExists: Optional[bool] = False):
"""
Registers the given :class:`~pyflink.table.TableDescriptor` as a
catalog table.
@@ -451,14 +456,17 @@ class TableEnvironment(object):
... .build())
... .option("rows-per-second", 10)
... .option("fields.f0.kind", "random")
- ... .build())
+ ... .build(),
+ ... True)
:param path: The path under which the table will be registered.
:param descriptor: Template for creating a 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.
.. versionadded:: 1.14.0
"""
- self._j_tenv.createTable(path, descriptor._j_table_descriptor)
+ self._j_tenv.createTable(path, descriptor._j_table_descriptor,
ignoreIfExists)
def from_path(self, path: str) -> Table:
"""
diff --git a/flink-python/pyflink/table/tests/test_table_environment_api.py
b/flink-python/pyflink/table/tests/test_table_environment_api.py
index e67a16b6cbd..5fe20d941a7 100644
--- a/flink-python/pyflink/table/tests/test_table_environment_api.py
+++ b/flink-python/pyflink/table/tests/test_table_environment_api.py
@@ -33,7 +33,7 @@ from pyflink.datastream.tests.test_util import
DataStreamTestSinkFunction
from pyflink.datastream.window import TimeWindowSerializer
from pyflink.java_gateway import get_gateway
from pyflink.table import (DataTypes, StreamTableEnvironment,
EnvironmentSettings, Module,
- ResultKind, ModuleEntry)
+ ResultKind, ModuleEntry, Schema)
from pyflink.table.catalog import ObjectPath, CatalogBaseTable
from pyflink.table.explain_detail import ExplainDetail
from pyflink.table.expressions import col, source_watermark
@@ -188,54 +188,70 @@ class TableEnvironmentTest(PyFlinkUTTestCase):
self.assert_equals(t_env.list_user_defined_functions(), [])
def test_create_temporary_table_from_descriptor(self):
- from pyflink.table.schema import Schema
+ schema = Schema.new_builder().column("f0", DataTypes.INT()).build()
- t_env = self.t_env
- catalog = t_env.get_current_catalog()
- database = t_env.get_current_database()
+ self.assert_created_temporary_table_from_descriptor(schema)
+
+ def test_create_temporary_table_if_not_exists_from_descriptor(self):
schema = Schema.new_builder().column("f0", DataTypes.INT()).build()
- t_env.create_temporary_table(
- "T",
- TableDescriptor.for_connector("fake")
- .schema(schema)
- .option("a", "Test")
- .build())
-
self.assertFalse(t_env.get_catalog(catalog).table_exists(ObjectPath(database,
"T")))
- gateway = get_gateway()
+ self.assert_created_temporary_table_from_descriptor(schema)
- catalog_table = CatalogBaseTable(
- t_env._j_tenv.getCatalogManager()
- .getTable(gateway.jvm.ObjectIdentifier.of(catalog, database,
"T"))
- .get()
- .getTable())
- self.assertEqual(schema, catalog_table.get_unresolved_schema())
- self.assertEqual("fake", catalog_table.get_options().get("connector"))
- self.assertEqual("Test", catalog_table.get_options().get("a"))
+ # This should be a no-op and throw no exception
+ self.t_env.create_temporary_table(
+ "T",
+ TableDescriptor.for_connector("fake")
+ .schema(schema)
+ .option("a", "Test")
+ .build(),
+ True)
+
+ with self.assertRaises(Exception) as error_context:
+ self.t_env.create_temporary_table(
+ "T",
+ TableDescriptor.for_connector("fake")
+ .schema(schema)
+ .option("a", "Test")
+ .build(),
+ False)
+
+ error_msg = str(error_context.exception)
+ self.assertIn("Temporary table
'`default_catalog`.`default_database`.`T`' already exists",
+ error_msg)
def test_create_table_from_descriptor(self):
- from pyflink.table.schema import Schema
+ schema = Schema.new_builder().column("f0", DataTypes.INT()).build()
- catalog = self.t_env.get_current_catalog()
- database = self.t_env.get_current_database()
+ self.assert_created_table_from_descriptor(schema)
+
+ def test_create_table_ignore_if_exists_from_descriptor(self):
schema = Schema.new_builder().column("f0", DataTypes.INT()).build()
+ self.assert_created_table_from_descriptor(schema)
+
+ # This should be a no-op and throw no exception
self.t_env.create_table(
"T",
TableDescriptor.for_connector("fake")
- .schema(schema)
- .option("a", "Test")
- .build())
- object_path = ObjectPath(database, "T")
-
self.assertTrue(self.t_env.get_catalog(catalog).table_exists(object_path))
-
- catalog_table = self.t_env.get_catalog(catalog).get_table(object_path)
- self.assertEqual(schema, catalog_table.get_unresolved_schema())
- self.assertEqual("fake", catalog_table.get_options().get("connector"))
- self.assertEqual("Test", catalog_table.get_options().get("a"))
+ .schema(schema)
+ .option("a", "Test")
+ .build(),
+ True)
+
+ with self.assertRaises(Exception) as error_context:
+ self.t_env.create_table(
+ "T",
+ TableDescriptor.for_connector("fake")
+ .schema(schema)
+ .option("a", "Test")
+ .build(),
+ False)
+
+ error_msg = str(error_context.exception)
+ self.assertIn(
+ "Could not execute CreateTable in path
`default_catalog`.`default_database`.`T`",
+ error_msg)
def test_table_from_descriptor(self):
- from pyflink.table.schema import Schema
-
schema = Schema.new_builder().column("f0", DataTypes.INT()).build()
descriptor =
TableDescriptor.for_connector("fake").schema(schema).build()
@@ -326,6 +342,48 @@ class TableEnvironmentTest(PyFlinkUTTestCase):
Py4JJavaError, "No module with name 'dummy' exists",
self.t_env.use_modules, 'core', 'dummy')
+ def assert_created_table_from_descriptor(self, schema: Schema):
+ catalog = self.t_env.get_current_catalog()
+ database = self.t_env.get_current_database()
+
+ self.t_env.create_table(
+ "T",
+ TableDescriptor.for_connector("fake")
+ .schema(schema)
+ .option("a", "Test")
+ .build())
+
+ object_path = ObjectPath(database, "T")
+
self.assertTrue(self.t_env.get_catalog(catalog).table_exists(object_path))
+
+ catalog_table = self.t_env.get_catalog(catalog).get_table(object_path)
+ self.assertEqual(schema, catalog_table.get_unresolved_schema())
+ self.assertEqual("fake", catalog_table.get_options().get("connector"))
+ self.assertEqual("Test", catalog_table.get_options().get("a"))
+
+ def assert_created_temporary_table_from_descriptor(self, schema: Schema):
+ catalog = self.t_env.get_current_catalog()
+ database = self.t_env.get_current_database()
+
+ self.t_env.create_temporary_table(
+ "T",
+ TableDescriptor.for_connector("fake")
+ .schema(schema)
+ .option("a", "Test")
+ .build())
+
+
self.assertFalse(self.t_env.get_catalog(catalog).table_exists(ObjectPath(database,
"T")))
+ gateway = get_gateway()
+
+ catalog_table = CatalogBaseTable(
+ self.t_env._j_tenv.getCatalogManager()
+ .getTable(gateway.jvm.ObjectIdentifier.of(catalog, database, "T"))
+ .get()
+ .getTable())
+ self.assertEqual(schema, catalog_table.get_unresolved_schema())
+ self.assertEqual("fake", catalog_table.get_options().get("connector"))
+ self.assertEqual("Test", catalog_table.get_options().get("a"))
+
class DataStreamConversionTestCases(PyFlinkUTTestCase):
@@ -415,8 +473,6 @@ class DataStreamConversionTestCases(PyFlinkUTTestCase):
self.assert_equals(result, expected)
def test_from_data_stream_with_schema(self):
- from pyflink.table import Schema
-
ds = self.env.from_collection([(1, 'Hi', 'Hello'), (2, 'Hello', 'Hi')],
type_info=Types.ROW_NAMED(
["a", "b", "c"],
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 bc1ec5bc255..c54722f5dd6 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
@@ -718,6 +718,37 @@ public interface TableEnvironment {
*/
void createTemporaryTable(String path, TableDescriptor descriptor);
+ /**
+ * Registers the given {@link TableDescriptor} as a temporary catalog
table.
+ *
+ * <p>The {@link TableDescriptor descriptor} is converted into a {@link
CatalogTable} and stored
+ * in the catalog.
+ *
+ * <p>Temporary objects can shadow permanent ones. If a permanent object
in a given path exists,
+ * it will be inaccessible in the current session. To make the permanent
object available again
+ * one can drop the corresponding temporary object.
+ *
+ * <p>Examples:
+ *
+ * <pre>{@code
+ * tEnv.createTemporaryTable("MyTable",
TableDescriptor.forConnector("datagen")
+ * .schema(Schema.newBuilder()
+ * .column("f0", DataTypes.STRING())
+ * .build())
+ * .option(DataGenOptions.ROWS_PER_SECOND, 10)
+ * .option("fields.f0.kind", "random")
+ * .build(),
+ * true);
+ * }</pre>
+ *
+ * @param path The path under which the table will be registered. See also
the {@link
+ * TableEnvironment} class description for the format of the path.
+ * @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.
+ */
+ void createTemporaryTable(String path, TableDescriptor descriptor, boolean
ignoreIfExists);
+
/**
* Registers the given {@link TableDescriptor} as a catalog table.
*
@@ -745,6 +776,36 @@ public interface TableEnvironment {
*/
void createTable(String path, TableDescriptor descriptor);
+ /**
+ * Registers the given {@link TableDescriptor} as a catalog table.
+ *
+ * <p>The {@link TableDescriptor descriptor} is converted into a {@link
CatalogTable} and stored
+ * in the catalog.
+ *
+ * <p>If the table should not be permanently stored in a catalog, use
{@link
+ * #createTemporaryTable(String, TableDescriptor, boolean)} instead.
+ *
+ * <p>Examples:
+ *
+ * <pre>{@code
+ * tEnv.createTable("MyTable", TableDescriptor.forConnector("datagen")
+ * .schema(Schema.newBuilder()
+ * .column("f0", DataTypes.STRING())
+ * .build())
+ * .option(DataGenOptions.ROWS_PER_SECOND, 10)
+ * .option("fields.f0.kind", "random")
+ * .build(),
+ * true);
+ * }</pre>
+ *
+ * @param path The path under which the table will be registered. See also
the {@link
+ * TableEnvironment} class description for the format of the path.
+ * @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.
+ */
+ void createTable(String path, TableDescriptor descriptor, boolean
ignoreIfExists);
+
/**
* Registers a {@link Table} under a unique name in the TableEnvironment's
catalog. Registered
* tables can be referenced in SQL queries.
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 b801bc4aaea..f3536ce4385 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
@@ -485,22 +485,34 @@ public class TableEnvironmentImpl implements
TableEnvironmentInternal {
@Override
public void createTemporaryTable(String path, TableDescriptor descriptor) {
+ this.createTemporaryTable(path, descriptor, false);
+ }
+
+ @Override
+ public void createTemporaryTable(
+ 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.createTemporaryTable(descriptor.toCatalogTable(),
tableIdentifier, false);
+ catalogManager.createTemporaryTable(
+ descriptor.toCatalogTable(), tableIdentifier, ignoreIfExists);
}
@Override
public void createTable(String path, TableDescriptor descriptor) {
+ this.createTable(path, descriptor, false);
+ }
+
+ @Override
+ public void 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, false);
+ catalogManager.createTable(descriptor.toCatalogTable(),
tableIdentifier, ignoreIfExists);
}
@Override
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 9c7fd8c2c03..b449fce3f35 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
@@ -39,79 +39,66 @@ import static
org.apache.flink.table.factories.TestManagedTableFactory.ENRICHED_
import static
org.apache.flink.table.factories.TestManagedTableFactory.ENRICHED_VALUE;
import static
org.apache.flink.table.factories.TestManagedTableFactory.MANAGED_TABLES;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatNoException;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.assertj.core.api.Assertions.entry;
import static org.assertj.core.api.InstanceOfAssertFactories.type;
/** Tests for {@link TableEnvironment}. */
class TableEnvironmentTest {
+ private static final Schema TEST_SCHEMA =
+ Schema.newBuilder().column("f0", DataTypes.INT()).build();
+ private static final TableDescriptor TEST_DESCRIPTOR =
+
TableDescriptor.forConnector("fake").schema(TEST_SCHEMA).option("a",
"Test").build();
@Test
void testCreateTemporaryTableFromDescriptor() {
final TableEnvironmentMock tEnv =
TableEnvironmentMock.getStreamingInstance();
- final String catalog = tEnv.getCurrentCatalog();
- final String database = tEnv.getCurrentDatabase();
- final Schema schema = Schema.newBuilder().column("f0",
DataTypes.INT()).build();
- tEnv.createTemporaryTable(
- "T",
-
TableDescriptor.forConnector("fake").schema(schema).option("a",
"Test").build());
+ assertTemporaryCreateTableFromDescriptor(tEnv, TEST_SCHEMA, false);
+ }
- assertThat(
- tEnv.getCatalog(catalog)
- .orElseThrow(AssertionError::new)
- .tableExists(new ObjectPath(database, "T")))
- .isFalse();
+ @Test
+ void testCreateTemporaryTableIfNotExistsFromDescriptor() {
+ final TableEnvironmentMock tEnv =
TableEnvironmentMock.getStreamingInstance();
- final Optional<ContextResolvedTable> lookupResult =
- tEnv.getCatalogManager().getTable(ObjectIdentifier.of(catalog,
database, "T"));
- assertThat(lookupResult.isPresent()).isTrue();
+ assertTemporaryCreateTableFromDescriptor(tEnv, TEST_SCHEMA, true);
+ assertThatNoException()
+ .isThrownBy(() -> tEnv.createTemporaryTable("T",
TEST_DESCRIPTOR, true));
- final CatalogBaseTable catalogTable =
lookupResult.get().getResolvedTable();
- assertThat(catalogTable instanceof CatalogTable).isTrue();
- assertThat(catalogTable.getUnresolvedSchema()).isEqualTo(schema);
-
assertThat(catalogTable.getOptions().get("connector")).isEqualTo("fake");
- assertThat(catalogTable.getOptions().get("a")).isEqualTo("Test");
+ assertThatThrownBy(() -> tEnv.createTemporaryTable("T",
TEST_DESCRIPTOR, false))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining(
+ "Temporary table
'`default_catalog`.`default_database`.`T`' already exists");
}
@Test
void testCreateTableFromDescriptor() throws Exception {
final TableEnvironmentMock tEnv =
TableEnvironmentMock.getStreamingInstance();
- final String catalog = tEnv.getCurrentCatalog();
- final String database = tEnv.getCurrentDatabase();
- final Schema schema = Schema.newBuilder().column("f0",
DataTypes.INT()).build();
- tEnv.createTable(
- "T",
-
TableDescriptor.forConnector("fake").schema(schema).option("a",
"Test").build());
+ assertCreateTableFromDescriptor(tEnv, TEST_SCHEMA, false);
+ }
- final ObjectPath objectPath = new ObjectPath(database, "T");
- assertThat(
- tEnv.getCatalog(catalog)
- .orElseThrow(AssertionError::new)
- .tableExists(objectPath))
- .isTrue();
+ @Test
+ void testCreateTableIfNotExistsFromDescriptor() throws Exception {
+ final TableEnvironmentMock tEnv =
TableEnvironmentMock.getStreamingInstance();
- final CatalogBaseTable catalogTable =
-
tEnv.getCatalog(catalog).orElseThrow(AssertionError::new).getTable(objectPath);
- assertThat(catalogTable).isInstanceOf(CatalogTable.class);
- assertThat(catalogTable.getUnresolvedSchema()).isEqualTo(schema);
- assertThat(catalogTable.getOptions())
- .contains(entry("connector", "fake"), entry("a", "Test"));
+ assertCreateTableFromDescriptor(tEnv, TEST_SCHEMA, true);
+ assertThatNoException().isThrownBy(() -> tEnv.createTable("T",
TEST_DESCRIPTOR, true));
+
+ assertThatThrownBy(() -> tEnv.createTable("T", TEST_DESCRIPTOR, false))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining(
+ "Could not execute CreateTable in path
`default_catalog`.`default_database`.`T`");
}
@Test
void testTableFromDescriptor() {
final TableEnvironmentMock tEnv =
TableEnvironmentMock.getStreamingInstance();
-
- final Schema schema = Schema.newBuilder().column("f0",
DataTypes.INT()).build();
- final TableDescriptor descriptor =
- TableDescriptor.forConnector("fake").schema(schema).build();
-
- final Table table = tEnv.from(descriptor);
+ final Table table = tEnv.from(TEST_DESCRIPTOR);
assertThat(Schema.newBuilder().fromResolvedSchema(table.getResolvedSchema()).build())
- .isEqualTo(schema);
+ .isEqualTo(TEST_SCHEMA);
assertThat(table.getQueryOperation())
.asInstanceOf(type(SourceQueryOperation.class))
@@ -147,12 +134,66 @@ class TableEnvironmentTest {
innerTestManagedTableFromDescriptor(true, true);
}
+ private static void assertCreateTableFromDescriptor(
+ TableEnvironmentMock tEnv, Schema schema, boolean ignoreIfExists)
+ throws
org.apache.flink.table.catalog.exceptions.TableNotExistException {
+ final String catalog = tEnv.getCurrentCatalog();
+ final String database = tEnv.getCurrentDatabase();
+
+ if (ignoreIfExists) {
+ tEnv.createTable("T", TEST_DESCRIPTOR, true);
+ } else {
+ tEnv.createTable("T", TEST_DESCRIPTOR);
+ }
+
+ final ObjectPath objectPath = new ObjectPath(database, "T");
+ assertThat(
+ tEnv.getCatalog(catalog)
+ .orElseThrow(AssertionError::new)
+ .tableExists(objectPath))
+ .isTrue();
+
+ final CatalogBaseTable catalogTable =
+
tEnv.getCatalog(catalog).orElseThrow(AssertionError::new).getTable(objectPath);
+ assertThat(catalogTable).isInstanceOf(CatalogTable.class);
+ assertThat(catalogTable.getUnresolvedSchema()).isEqualTo(schema);
+ assertThat(catalogTable.getOptions())
+ .contains(entry("connector", "fake"), entry("a", "Test"));
+ }
+
+ private static void assertTemporaryCreateTableFromDescriptor(
+ TableEnvironmentMock tEnv, Schema schema, boolean ignoreIfExists) {
+ final String catalog = tEnv.getCurrentCatalog();
+ final String database = tEnv.getCurrentDatabase();
+
+ if (ignoreIfExists) {
+ tEnv.createTemporaryTable("T", TEST_DESCRIPTOR, true);
+ } else {
+ tEnv.createTemporaryTable("T", TEST_DESCRIPTOR);
+ }
+
+ assertThat(
+ tEnv.getCatalog(catalog)
+ .orElseThrow(AssertionError::new)
+ .tableExists(new ObjectPath(database, "T")))
+ .isFalse();
+
+ final Optional<ContextResolvedTable> lookupResult =
+ tEnv.getCatalogManager().getTable(ObjectIdentifier.of(catalog,
database, "T"));
+ assertThat(lookupResult.isPresent()).isTrue();
+
+ final CatalogBaseTable catalogTable =
lookupResult.get().getResolvedTable();
+ assertThat(catalogTable instanceof CatalogTable).isTrue();
+ assertThat(catalogTable.getUnresolvedSchema()).isEqualTo(schema);
+
assertThat(catalogTable.getOptions().get("connector")).isEqualTo("fake");
+ assertThat(catalogTable.getOptions().get("a")).isEqualTo("Test");
+ }
+
private void innerTestManagedTableFromDescriptor(boolean ignoreIfExists,
boolean isTemporary) {
final TableEnvironmentMock tEnv =
TableEnvironmentMock.getStreamingInstance();
final String catalog = tEnv.getCurrentCatalog();
final String database = tEnv.getCurrentDatabase();
- final Schema schema = Schema.newBuilder().column("f0",
DataTypes.INT()).build();
final String tableName = UUID.randomUUID().toString();
ObjectIdentifier identifier = ObjectIdentifier.of(catalog, database,
tableName);
@@ -162,7 +203,7 @@ class TableEnvironmentTest {
new CreateTableOperation(
identifier,
TableDescriptor.forManaged()
- .schema(schema)
+ .schema(TEST_SCHEMA)
.option("a", "Test")
.build()
.toCatalogTable(),
@@ -198,7 +239,7 @@ class TableEnvironmentTest {
final CatalogBaseTable catalogTable =
lookupResult.get().getResolvedTable();
assertThat(catalogTable instanceof CatalogTable).isTrue();
- assertThat(catalogTable.getUnresolvedSchema()).isEqualTo(schema);
+ assertThat(catalogTable.getUnresolvedSchema()).isEqualTo(TEST_SCHEMA);
assertThat(catalogTable.getOptions().get("a")).isEqualTo("Test");
assertThat(catalogTable.getOptions().get(ENRICHED_KEY)).isEqualTo(ENRICHED_VALUE);
