You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "lsyldliu (via GitHub)" <gi...@apache.org> on 2024/04/24 11:37:32 UTC
[PR] [FLINK-35189][testing/test-filesystem]Introduce test-filesystem Catalog based on FileSystem Connector to support materialized table [flink]
lsyldliu opened a new pull request, #24712:
URL: https://github.com/apache/flink/pull/24712
## What is the purpose of the change
Introduce test-filesystem Catalog and Connector based on FileSystem Connector to support materialized table.
## Brief change log
- *CatalogPropertiesUtil support serialize and deserialize materialized table*
- *Modify the visibility of filesystem connector-related methods to protected *
- *Introduce test-filesystem connector and catalog based on filesystem to support materialized table*
## Verifying this change
This change added tests and can be verified as follows:
- *Added tests: TestFileSystemCatalogFactoryTest, TestFileSystemCatalogITCase, TestFileSystemCatalogTest, TestFileSystemTableFactoryTest*
## 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)`: (no)
- The serializers: (no)
- The runtime per-record code paths (performance sensitive): (no)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, 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? (JavaDocs)
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35189][testing/test-filesystem]Introduce test-filesystem Catalog based on FileSystem Connector to support materialized table [flink]
Posted by "lsyldliu (via GitHub)" <gi...@apache.org>.
lsyldliu commented on code in PR #24712:
URL: https://github.com/apache/flink/pull/24712#discussion_r1580355919
##########
flink-test-utils-parent/flink-table-filesystem-test-utils/src/main/java/org/apache/flink/table/file/testutils/catalog/TestFileSystemCatalog.java:
##########
@@ -0,0 +1,687 @@
+/*
+ * 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.file.testutils.catalog;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.java.tuple.Tuple4;
+import org.apache.flink.connector.file.table.FileSystemConnectorOptions;
+import org.apache.flink.core.fs.FSDataInputStream;
+import org.apache.flink.core.fs.FSDataOutputStream;
+import org.apache.flink.core.fs.FileStatus;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.table.catalog.AbstractCatalog;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogDatabase;
+import org.apache.flink.table.catalog.CatalogDatabaseImpl;
+import org.apache.flink.table.catalog.CatalogFunction;
+import org.apache.flink.table.catalog.CatalogMaterializedTable;
+import org.apache.flink.table.catalog.CatalogPartition;
+import org.apache.flink.table.catalog.CatalogPartitionSpec;
+import org.apache.flink.table.catalog.CatalogPropertiesUtil;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.CatalogView;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.ResolvedCatalogBaseTable;
+import org.apache.flink.table.catalog.ResolvedCatalogMaterializedTable;
+import org.apache.flink.table.catalog.ResolvedCatalogTable;
+import org.apache.flink.table.catalog.TableChange;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.DatabaseNotEmptyException;
+import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
+import org.apache.flink.table.catalog.exceptions.FunctionAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.FunctionNotExistException;
+import org.apache.flink.table.catalog.exceptions.PartitionAlreadyExistsException;
+import org.apache.flink.table.catalog.exceptions.PartitionNotExistException;
+import org.apache.flink.table.catalog.exceptions.PartitionSpecInvalidException;
+import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.TableNotExistException;
+import org.apache.flink.table.catalog.exceptions.TableNotPartitionedException;
+import org.apache.flink.table.catalog.exceptions.TablePartitionedException;
+import org.apache.flink.table.catalog.stats.CatalogColumnStatistics;
+import org.apache.flink.table.catalog.stats.CatalogTableStatistics;
+import org.apache.flink.table.expressions.Expression;
+import org.apache.flink.table.factories.Factory;
+import org.apache.flink.table.file.testutils.TestFileSystemTableFactory;
+import org.apache.flink.util.CollectionUtil;
+import org.apache.flink.util.Preconditions;
+import org.apache.flink.util.StringUtils;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+/** A catalog implementation for test {@link FileSystem}. */
+public class TestFileSystemCatalog extends AbstractCatalog {
+
+ public static final String SCHEMA_PATH = "schema";
+ public static final String DATA_PATH = "data";
+ private static final String SCHEMA_FILE_EXTENSION = ".json";
+
+ private final String catalogPathStr;
+
+ private Path catalogPath;
+ private FileSystem fs;
+
+ public TestFileSystemCatalog(String pathStr, String name, String defaultDatabase) {
+ super(name, defaultDatabase);
+ this.catalogPathStr = pathStr;
+ }
+
+ @VisibleForTesting
+ public String getCatalogPathStr() {
+ return catalogPathStr;
+ }
+
+ @Override
+ public Optional<Factory> getFactory() {
+ return Optional.of(new TestFileSystemTableFactory());
+ }
+
+ @Override
+ public void open() throws CatalogException {
+ try {
+ catalogPath = new Path(catalogPathStr);
+ fs = catalogPath.getFileSystem();
+ if (!fs.exists(catalogPath)) {
+ throw new CatalogException(
+ String.format(
+ "Catalog %s path %s does not exist.", getName(), catalogPath));
+ }
+ if (!fs.getFileStatus(catalogPath).isDir()) {
+ throw new CatalogException(
+ String.format(
+ "Failed to open catalog path. The given path %s is not a directory.",
+ catalogPath));
+ }
+ } catch (IOException e) {
+ throw new CatalogException(
+ String.format("Checking catalog path %s exists occur exception.", catalogPath),
+ e);
+ }
+ }
+
+ @Override
+ public void close() throws CatalogException {}
+
+ @Override
+ public List<String> listDatabases() throws CatalogException {
+ try {
+ FileStatus[] fileStatuses = fs.listStatus(catalogPath);
+ return Arrays.stream(fileStatuses)
+ .filter(FileStatus::isDir)
+ .map(fileStatus -> fileStatus.getPath().getName())
+ .collect(Collectors.toList());
+ } catch (IOException e) {
+ throw new CatalogException("Listing database occur exception.", e);
+ }
+ }
+
+ @Override
+ public CatalogDatabase getDatabase(String databaseName)
+ throws DatabaseNotExistException, CatalogException {
+ if (databaseExists(databaseName)) {
+ return new CatalogDatabaseImpl(Collections.emptyMap(), null);
+ } else {
+ throw new DatabaseNotExistException(getName(), databaseName);
+ }
+ }
+
+ @Override
+ public boolean databaseExists(String databaseName) throws CatalogException {
+ Preconditions.checkArgument(!StringUtils.isNullOrWhitespaceOnly(databaseName));
+
+ return listDatabases().contains(databaseName);
+ }
+
+ @Override
+ public void createDatabase(String name, CatalogDatabase database, boolean ignoreIfExists)
+ throws DatabaseAlreadyExistException, CatalogException {
+ if (databaseExists(name)) {
Review Comment:
I think there is no need, we should trust the users and leave the choice to them. There may be other unsupported patterns for database names, so we can't enumerate and restrict them.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35189][testing/test-filesystem]Introduce test-filesystem Catalog based on FileSystem Connector to support materialized table [flink]
Posted by "hackergin (via GitHub)" <gi...@apache.org>.
hackergin commented on code in PR #24712:
URL: https://github.com/apache/flink/pull/24712#discussion_r1579942461
##########
flink-test-utils-parent/flink-table-filesystem-test-utils/src/main/java/org/apache/flink/table/file/testutils/catalog/TestFileSystemCatalog.java:
##########
@@ -0,0 +1,687 @@
+/*
+ * 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.file.testutils.catalog;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.java.tuple.Tuple4;
+import org.apache.flink.connector.file.table.FileSystemConnectorOptions;
+import org.apache.flink.core.fs.FSDataInputStream;
+import org.apache.flink.core.fs.FSDataOutputStream;
+import org.apache.flink.core.fs.FileStatus;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.table.catalog.AbstractCatalog;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogDatabase;
+import org.apache.flink.table.catalog.CatalogDatabaseImpl;
+import org.apache.flink.table.catalog.CatalogFunction;
+import org.apache.flink.table.catalog.CatalogMaterializedTable;
+import org.apache.flink.table.catalog.CatalogPartition;
+import org.apache.flink.table.catalog.CatalogPartitionSpec;
+import org.apache.flink.table.catalog.CatalogPropertiesUtil;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.CatalogView;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.ResolvedCatalogBaseTable;
+import org.apache.flink.table.catalog.ResolvedCatalogMaterializedTable;
+import org.apache.flink.table.catalog.ResolvedCatalogTable;
+import org.apache.flink.table.catalog.TableChange;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.DatabaseNotEmptyException;
+import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
+import org.apache.flink.table.catalog.exceptions.FunctionAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.FunctionNotExistException;
+import org.apache.flink.table.catalog.exceptions.PartitionAlreadyExistsException;
+import org.apache.flink.table.catalog.exceptions.PartitionNotExistException;
+import org.apache.flink.table.catalog.exceptions.PartitionSpecInvalidException;
+import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.TableNotExistException;
+import org.apache.flink.table.catalog.exceptions.TableNotPartitionedException;
+import org.apache.flink.table.catalog.exceptions.TablePartitionedException;
+import org.apache.flink.table.catalog.stats.CatalogColumnStatistics;
+import org.apache.flink.table.catalog.stats.CatalogTableStatistics;
+import org.apache.flink.table.expressions.Expression;
+import org.apache.flink.table.factories.Factory;
+import org.apache.flink.table.file.testutils.TestFileSystemTableFactory;
+import org.apache.flink.util.CollectionUtil;
+import org.apache.flink.util.Preconditions;
+import org.apache.flink.util.StringUtils;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+/** A catalog implementation for test {@link FileSystem}. */
+public class TestFileSystemCatalog extends AbstractCatalog {
+
+ public static final String SCHEMA_PATH = "schema";
+ public static final String DATA_PATH = "data";
+ private static final String SCHEMA_FILE_EXTENSION = ".json";
+
+ private final String catalogPathStr;
+
+ private Path catalogPath;
+ private FileSystem fs;
+
+ public TestFileSystemCatalog(String pathStr, String name, String defaultDatabase) {
+ super(name, defaultDatabase);
+ this.catalogPathStr = pathStr;
+ }
+
+ @VisibleForTesting
+ public String getCatalogPathStr() {
+ return catalogPathStr;
+ }
+
+ @Override
+ public Optional<Factory> getFactory() {
+ return Optional.of(new TestFileSystemTableFactory());
+ }
+
+ @Override
+ public void open() throws CatalogException {
+ try {
+ catalogPath = new Path(catalogPathStr);
+ fs = catalogPath.getFileSystem();
+ if (!fs.exists(catalogPath)) {
+ throw new CatalogException(
+ String.format(
+ "Catalog %s path %s does not exist.", getName(), catalogPath));
+ }
+ if (!fs.getFileStatus(catalogPath).isDir()) {
+ throw new CatalogException(
+ String.format(
+ "Failed to open catalog path. The given path %s is not a directory.",
+ catalogPath));
+ }
+ } catch (IOException e) {
+ throw new CatalogException(
+ String.format("Checking catalog path %s exists occur exception.", catalogPath),
+ e);
+ }
+ }
+
+ @Override
+ public void close() throws CatalogException {}
+
+ @Override
+ public List<String> listDatabases() throws CatalogException {
+ try {
+ FileStatus[] fileStatuses = fs.listStatus(catalogPath);
+ return Arrays.stream(fileStatuses)
+ .filter(FileStatus::isDir)
+ .map(fileStatus -> fileStatus.getPath().getName())
+ .collect(Collectors.toList());
+ } catch (IOException e) {
+ throw new CatalogException("Listing database occur exception.", e);
+ }
+ }
+
+ @Override
+ public CatalogDatabase getDatabase(String databaseName)
+ throws DatabaseNotExistException, CatalogException {
+ if (databaseExists(databaseName)) {
+ return new CatalogDatabaseImpl(Collections.emptyMap(), null);
+ } else {
+ throw new DatabaseNotExistException(getName(), databaseName);
+ }
+ }
+
+ @Override
+ public boolean databaseExists(String databaseName) throws CatalogException {
+ Preconditions.checkArgument(!StringUtils.isNullOrWhitespaceOnly(databaseName));
Review Comment:
Here can we add a error message here? The database name cannot be null or empty.
##########
flink-test-utils-parent/flink-table-filesystem-test-utils/src/main/java/org/apache/flink/table/file/testutils/catalog/TestFileSystemCatalog.java:
##########
@@ -0,0 +1,687 @@
+/*
+ * 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.file.testutils.catalog;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.java.tuple.Tuple4;
+import org.apache.flink.connector.file.table.FileSystemConnectorOptions;
+import org.apache.flink.core.fs.FSDataInputStream;
+import org.apache.flink.core.fs.FSDataOutputStream;
+import org.apache.flink.core.fs.FileStatus;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.table.catalog.AbstractCatalog;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogDatabase;
+import org.apache.flink.table.catalog.CatalogDatabaseImpl;
+import org.apache.flink.table.catalog.CatalogFunction;
+import org.apache.flink.table.catalog.CatalogMaterializedTable;
+import org.apache.flink.table.catalog.CatalogPartition;
+import org.apache.flink.table.catalog.CatalogPartitionSpec;
+import org.apache.flink.table.catalog.CatalogPropertiesUtil;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.CatalogView;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.ResolvedCatalogBaseTable;
+import org.apache.flink.table.catalog.ResolvedCatalogMaterializedTable;
+import org.apache.flink.table.catalog.ResolvedCatalogTable;
+import org.apache.flink.table.catalog.TableChange;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.DatabaseNotEmptyException;
+import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
+import org.apache.flink.table.catalog.exceptions.FunctionAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.FunctionNotExistException;
+import org.apache.flink.table.catalog.exceptions.PartitionAlreadyExistsException;
+import org.apache.flink.table.catalog.exceptions.PartitionNotExistException;
+import org.apache.flink.table.catalog.exceptions.PartitionSpecInvalidException;
+import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.TableNotExistException;
+import org.apache.flink.table.catalog.exceptions.TableNotPartitionedException;
+import org.apache.flink.table.catalog.exceptions.TablePartitionedException;
+import org.apache.flink.table.catalog.stats.CatalogColumnStatistics;
+import org.apache.flink.table.catalog.stats.CatalogTableStatistics;
+import org.apache.flink.table.expressions.Expression;
+import org.apache.flink.table.factories.Factory;
+import org.apache.flink.table.file.testutils.TestFileSystemTableFactory;
+import org.apache.flink.util.CollectionUtil;
+import org.apache.flink.util.Preconditions;
+import org.apache.flink.util.StringUtils;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+/** A catalog implementation for test {@link FileSystem}. */
+public class TestFileSystemCatalog extends AbstractCatalog {
+
+ public static final String SCHEMA_PATH = "schema";
+ public static final String DATA_PATH = "data";
+ private static final String SCHEMA_FILE_EXTENSION = ".json";
+
+ private final String catalogPathStr;
+
+ private Path catalogPath;
+ private FileSystem fs;
+
+ public TestFileSystemCatalog(String pathStr, String name, String defaultDatabase) {
+ super(name, defaultDatabase);
+ this.catalogPathStr = pathStr;
+ }
+
+ @VisibleForTesting
+ public String getCatalogPathStr() {
+ return catalogPathStr;
+ }
+
+ @Override
+ public Optional<Factory> getFactory() {
+ return Optional.of(new TestFileSystemTableFactory());
+ }
+
+ @Override
+ public void open() throws CatalogException {
+ try {
+ catalogPath = new Path(catalogPathStr);
+ fs = catalogPath.getFileSystem();
+ if (!fs.exists(catalogPath)) {
+ throw new CatalogException(
+ String.format(
+ "Catalog %s path %s does not exist.", getName(), catalogPath));
+ }
+ if (!fs.getFileStatus(catalogPath).isDir()) {
+ throw new CatalogException(
+ String.format(
+ "Failed to open catalog path. The given path %s is not a directory.",
+ catalogPath));
+ }
+ } catch (IOException e) {
+ throw new CatalogException(
+ String.format("Checking catalog path %s exists occur exception.", catalogPath),
+ e);
+ }
+ }
+
+ @Override
+ public void close() throws CatalogException {}
+
+ @Override
+ public List<String> listDatabases() throws CatalogException {
+ try {
+ FileStatus[] fileStatuses = fs.listStatus(catalogPath);
+ return Arrays.stream(fileStatuses)
+ .filter(FileStatus::isDir)
+ .map(fileStatus -> fileStatus.getPath().getName())
+ .collect(Collectors.toList());
+ } catch (IOException e) {
+ throw new CatalogException("Listing database occur exception.", e);
+ }
+ }
+
+ @Override
+ public CatalogDatabase getDatabase(String databaseName)
+ throws DatabaseNotExistException, CatalogException {
+ if (databaseExists(databaseName)) {
+ return new CatalogDatabaseImpl(Collections.emptyMap(), null);
+ } else {
+ throw new DatabaseNotExistException(getName(), databaseName);
+ }
+ }
+
+ @Override
+ public boolean databaseExists(String databaseName) throws CatalogException {
+ Preconditions.checkArgument(!StringUtils.isNullOrWhitespaceOnly(databaseName));
+
+ return listDatabases().contains(databaseName);
+ }
+
+ @Override
+ public void createDatabase(String name, CatalogDatabase database, boolean ignoreIfExists)
+ throws DatabaseAlreadyExistException, CatalogException {
+ if (databaseExists(name)) {
Review Comment:
Do we need to check if the database name contains "/" like db1/db2/db3 ? Otherwise, it may create subdirectories here.
##########
flink-test-utils-parent/flink-table-filesystem-test-utils/src/test/java/org/apache/flink/table/file/testutils/catalog/TestFileSystemCatalogFactoryTest.java:
##########
@@ -0,0 +1,92 @@
+/*
+ * 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.file.testutils.catalog;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CollectionUtil;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link TestFileSystemCatalog} created by {@link TestFileSystemCatalogFactory}. */
+public class TestFileSystemCatalogFactoryTest {
+
+ private static final String TEST_CATALOG = "test_catalog";
+ private static final String TEST_DEFAULT_DATABASE = "test_db";
+
+ @TempDir File tempFile;
+
+ @BeforeEach
+ void before() {
+ File testDb = new File(tempFile, TEST_DEFAULT_DATABASE);
+ testDb.mkdir();
+ }
+
+ @Test
+ public void testCreateCatalogWithDDL() {
+ final TableEnvironment tEnv =
+ TableEnvironment.create(EnvironmentSettings.inStreamingMode());
+ String catalogDDL =
+ String.format(
+ "CREATE CATALOG %s\n"
+ + "WITH (\n"
+ + " 'type' = 'test-filesystem',\n"
+ + " 'path' = '%s',\n"
+ + " 'default-database' = '%s'\n"
+ + " )",
+ TEST_CATALOG, tempFile.getAbsolutePath(), TEST_DEFAULT_DATABASE);
+
+ tEnv.executeSql(catalogDDL);
+
+ List<Row> result =
+ CollectionUtil.iteratorToList(tEnv.executeSql("SHOW CATALOGS").collect());
+ assertThat(result.contains(Row.of(TEST_CATALOG))).isTrue();
+ }
+
+ @Test
+ public void testCreateCatalogWithFactory() {
+ final Map<String, String> properties = new HashMap<>();
+ properties.put("type", "test-filesystem");
+ properties.put("default-database", TEST_CATALOG);
+ properties.put("path", tempFile.getAbsolutePath());
+ final TestFileSystemCatalog actualCatalog =
+ (TestFileSystemCatalog)
+ FactoryUtil.createCatalog(
+ TEST_CATALOG,
+ properties,
+ new Configuration(),
+ TestFileSystemCatalogFactoryTest.class.getClassLoader());
+
+ assertThat(actualCatalog.getCatalogPathStr()).isEqualTo(tempFile.getAbsolutePath());
+ assertThat(actualCatalog.getName()).isEqualTo(TEST_CATALOG);
+ assertThat(actualCatalog.getDefaultDatabase()).isEqualTo(TEST_CATALOG);
Review Comment:
```suggestion
assertThat(actualCatalog.getDefaultDatabase()).isEqualTo(TEST_DEFAULT_DATABASE);
```
##########
flink-test-utils-parent/flink-table-filesystem-test-utils/src/test/java/org/apache/flink/table/file/testutils/catalog/TestFileSystemCatalogFactoryTest.java:
##########
@@ -0,0 +1,92 @@
+/*
+ * 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.file.testutils.catalog;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CollectionUtil;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link TestFileSystemCatalog} created by {@link TestFileSystemCatalogFactory}. */
+public class TestFileSystemCatalogFactoryTest {
+
+ private static final String TEST_CATALOG = "test_catalog";
+ private static final String TEST_DEFAULT_DATABASE = "test_db";
+
+ @TempDir File tempFile;
+
+ @BeforeEach
+ void before() {
+ File testDb = new File(tempFile, TEST_DEFAULT_DATABASE);
+ testDb.mkdir();
+ }
+
+ @Test
+ public void testCreateCatalogWithDDL() {
+ final TableEnvironment tEnv =
+ TableEnvironment.create(EnvironmentSettings.inStreamingMode());
+ String catalogDDL =
+ String.format(
+ "CREATE CATALOG %s\n"
+ + "WITH (\n"
+ + " 'type' = 'test-filesystem',\n"
+ + " 'path' = '%s',\n"
+ + " 'default-database' = '%s'\n"
+ + " )",
+ TEST_CATALOG, tempFile.getAbsolutePath(), TEST_DEFAULT_DATABASE);
+
+ tEnv.executeSql(catalogDDL);
+
+ List<Row> result =
+ CollectionUtil.iteratorToList(tEnv.executeSql("SHOW CATALOGS").collect());
+ assertThat(result.contains(Row.of(TEST_CATALOG))).isTrue();
+ }
+
+ @Test
+ public void testCreateCatalogWithFactory() {
+ final Map<String, String> properties = new HashMap<>();
+ properties.put("type", "test-filesystem");
+ properties.put("default-database", TEST_CATALOG);
Review Comment:
```suggestion
properties.put("default-database", TEST_DEFAULT_DATABASE);
```
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35189][test-utils]Introduce test-filesystem Catalog based on FileSystem Connector to support materialized table [flink]
Posted by "lsyldliu (via GitHub)" <gi...@apache.org>.
lsyldliu merged PR #24712:
URL: https://github.com/apache/flink/pull/24712
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-35189][testing/test-filesystem]Introduce test-filesystem Catalog based on FileSystem Connector to support materialized table [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #24712:
URL: https://github.com/apache/flink/pull/24712#issuecomment-2074749559
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "4bc5adf12ee9507d83e0414f95f24fac6321cfb0",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "4bc5adf12ee9507d83e0414f95f24fac6321cfb0",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4bc5adf12ee9507d83e0414f95f24fac6321cfb0 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org