You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ro...@apache.org on 2024/04/26 09:41:54 UTC
(flink) branch master updated (f64654e9749 -> 714d1cb2e0b)
This is an automated email from the ASF dual-hosted git repository.
ron pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
from f64654e9749 add HTTP options to java-storage client
new 00d69e53fe7 [FLINK-35189][table-common] CatalogPropertiesUtil support serialize and deserialize materialized table
new e3cda01cac7 [FLINK-35189][connectors/filesystem] Modify the visibility of filesystem connector related methods to protected
new 714d1cb2e0b [FLINK-35189][test-utils] Introduce test-filesystem connector and catalog based on filesystem to support materialized table
The 3 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:
.../connector/file/src/AbstractFileSource.java | 11 +-
.../file/table/FileSystemTableFactory.java | 4 +-
.../file/table/FileSystemTableSource.java | 4 +-
.../flink/table/catalog/CatalogPropertiesUtil.java | 140 ++++-
.../flink-table-filesystem-test-utils/pom.xml | 103 +++
.../flink/connector/file/src/TestFileSource.java | 189 ++++++
.../file/table/TestFileSystemTableSource.java | 88 +++
.../file/testutils/TestFileSystemTableFactory.java | 54 ++
.../file/testutils/catalog/JsonSerdeUtil.java | 61 ++
.../testutils/catalog/TestFileSystemCatalog.java | 690 +++++++++++++++++++++
.../catalog/TestFileSystemCatalogFactory.java | 78 +++
.../org.apache.flink.table.factories.Factory | 17 +
.../testutils/TestFileSystemTableFactoryTest.java | 64 ++
.../catalog/TestFileSystemCatalogFactoryTest.java | 92 +++
.../catalog/TestFileSystemCatalogITCase.java | 89 +++
.../catalog/TestFileSystemCatalogTest.java | 379 +++++++++++
.../catalog/TestFileSystemCatalogTestBase.java | 56 ++
.../src/test/resources/log4j2-test.properties | 0
flink-test-utils-parent/pom.xml | 1 +
19 files changed, 2111 insertions(+), 9 deletions(-)
create mode 100644 flink-test-utils-parent/flink-table-filesystem-test-utils/pom.xml
create mode 100644 flink-test-utils-parent/flink-table-filesystem-test-utils/src/main/java/org/apache/flink/connector/file/src/TestFileSource.java
create mode 100644 flink-test-utils-parent/flink-table-filesystem-test-utils/src/main/java/org/apache/flink/connector/file/table/TestFileSystemTableSource.java
create mode 100644 flink-test-utils-parent/flink-table-filesystem-test-utils/src/main/java/org/apache/flink/table/file/testutils/TestFileSystemTableFactory.java
create mode 100644 flink-test-utils-parent/flink-table-filesystem-test-utils/src/main/java/org/apache/flink/table/file/testutils/catalog/JsonSerdeUtil.java
create mode 100644 flink-test-utils-parent/flink-table-filesystem-test-utils/src/main/java/org/apache/flink/table/file/testutils/catalog/TestFileSystemCatalog.java
create mode 100644 flink-test-utils-parent/flink-table-filesystem-test-utils/src/main/java/org/apache/flink/table/file/testutils/catalog/TestFileSystemCatalogFactory.java
create mode 100644 flink-test-utils-parent/flink-table-filesystem-test-utils/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
create mode 100644 flink-test-utils-parent/flink-table-filesystem-test-utils/src/test/java/org/apache/flink/table/file/testutils/TestFileSystemTableFactoryTest.java
create mode 100644 flink-test-utils-parent/flink-table-filesystem-test-utils/src/test/java/org/apache/flink/table/file/testutils/catalog/TestFileSystemCatalogFactoryTest.java
create mode 100644 flink-test-utils-parent/flink-table-filesystem-test-utils/src/test/java/org/apache/flink/table/file/testutils/catalog/TestFileSystemCatalogITCase.java
create mode 100644 flink-test-utils-parent/flink-table-filesystem-test-utils/src/test/java/org/apache/flink/table/file/testutils/catalog/TestFileSystemCatalogTest.java
create mode 100644 flink-test-utils-parent/flink-table-filesystem-test-utils/src/test/java/org/apache/flink/table/file/testutils/catalog/TestFileSystemCatalogTestBase.java
copy {flink-table/flink-table-api-java => flink-test-utils-parent/flink-table-filesystem-test-utils}/src/test/resources/log4j2-test.properties (100%)
(flink) 02/03: [FLINK-35189][connectors/filesystem] Modify the visibility of filesystem connector related methods to protected
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ron pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit e3cda01cac737c97a104feae0e35da1eb91a8751
Author: fengli <ld...@163.com>
AuthorDate: Wed Apr 24 18:14:53 2024 +0800
[FLINK-35189][connectors/filesystem] Modify the visibility of filesystem connector related methods to protected
---
.../apache/flink/connector/file/src/AbstractFileSource.java | 11 ++++++++---
.../flink/connector/file/table/FileSystemTableFactory.java | 4 ++--
.../flink/connector/file/table/FileSystemTableSource.java | 4 ++--
3 files changed, 12 insertions(+), 7 deletions(-)
diff --git a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/AbstractFileSource.java b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/AbstractFileSource.java
index f4fb463e10e..b14d46b3f9c 100644
--- a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/AbstractFileSource.java
+++ b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/src/AbstractFileSource.java
@@ -147,7 +147,7 @@ public abstract class AbstractFileSource<T, SplitT extends FileSourceSplit>
throw new FlinkRuntimeException("Could not enumerate file splits", e);
}
- return createSplitEnumerator(enumContext, enumerator, splits, null);
+ return createSplitEnumerator(getBoundedness(), enumContext, enumerator, splits, null);
}
@Override
@@ -164,7 +164,11 @@ public abstract class AbstractFileSource<T, SplitT extends FileSourceSplit>
(Collection<FileSourceSplit>) checkpoint.getSplits();
return createSplitEnumerator(
- enumContext, enumerator, splits, checkpoint.getAlreadyProcessedPaths());
+ getBoundedness(),
+ enumContext,
+ enumerator,
+ splits,
+ checkpoint.getAlreadyProcessedPaths());
}
@Override
@@ -186,6 +190,7 @@ public abstract class AbstractFileSource<T, SplitT extends FileSourceSplit>
// ------------------------------------------------------------------------
private SplitEnumerator<SplitT, PendingSplitsCheckpoint<SplitT>> createSplitEnumerator(
+ Boundedness boundedness,
SplitEnumeratorContext<SplitT> context,
FileEnumerator enumerator,
Collection<FileSourceSplit> splits,
@@ -199,7 +204,7 @@ public abstract class AbstractFileSource<T, SplitT extends FileSourceSplit>
final FileSplitAssigner splitAssigner = assignerFactory.create(splits);
- if (continuousEnumerationSettings == null) {
+ if (Boundedness.BOUNDED == boundedness) {
// bounded case
return castGeneric(new StaticFileSplitEnumerator(fileSplitContext, splitAssigner));
} else {
diff --git a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableFactory.java b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableFactory.java
index 3f0f415b863..16593ffd3e7 100644
--- a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableFactory.java
+++ b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableFactory.java
@@ -149,7 +149,7 @@ public class FileSystemTableFactory implements DynamicTableSourceFactory, Dynami
.collect(Collectors.toSet());
}
- private void validate(FactoryUtil.TableFactoryHelper helper) {
+ protected void validate(FactoryUtil.TableFactoryHelper helper) {
// Except format options, some formats like parquet and orc can not list all supported
// options.
helper.validateExcept(helper.getOptions().get(FactoryUtil.FORMAT) + ".");
@@ -160,7 +160,7 @@ public class FileSystemTableFactory implements DynamicTableSourceFactory, Dynami
.get(FileSystemConnectorOptions.SINK_PARTITION_COMMIT_WATERMARK_TIME_ZONE));
}
- private <I, F extends DecodingFormatFactory<I>> DecodingFormat<I> discoverDecodingFormat(
+ protected <I, F extends DecodingFormatFactory<I>> DecodingFormat<I> discoverDecodingFormat(
Context context, Class<F> formatFactoryClass) {
FactoryUtil.TableFactoryHelper helper = FactoryUtil.createTableFactoryHelper(this, context);
if (formatFactoryExists(context, formatFactoryClass)) {
diff --git a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableSource.java b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableSource.java
index 1e53eb53cfb..bbaa2a310d5 100644
--- a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableSource.java
+++ b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableSource.java
@@ -264,7 +264,7 @@ public class FileSystemTableSource extends AbstractFileSystemTable
return bulkFormat;
}
- private SourceProvider createSourceProvider(BulkFormat<RowData, FileSourceSplit> bulkFormat) {
+ protected SourceProvider createSourceProvider(BulkFormat<RowData, FileSourceSplit> bulkFormat) {
final FileSource.FileSourceBuilder<RowData> fileSourceBuilder =
FileSource.forBulkFileFormat(bulkFormat, paths());
@@ -287,7 +287,7 @@ public class FileSystemTableSource extends AbstractFileSystemTable
return SourceProvider.of(fileSourceBuilder.build());
}
- private Path[] paths() {
+ protected Path[] paths() {
if (partitionKeys.isEmpty()) {
return new Path[] {path};
} else {
(flink) 03/03: [FLINK-35189][test-utils] Introduce test-filesystem connector and catalog based on filesystem to support materialized table
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ron pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 714d1cb2e0bd0df03393492dc87cbd800af63e1b
Author: fengli <ld...@163.com>
AuthorDate: Wed Apr 24 18:18:47 2024 +0800
[FLINK-35189][test-utils] Introduce test-filesystem connector and catalog based on filesystem to support materialized table
---
.../flink-table-filesystem-test-utils/pom.xml | 103 +++
.../flink/connector/file/src/TestFileSource.java | 189 ++++++
.../file/table/TestFileSystemTableSource.java | 88 +++
.../file/testutils/TestFileSystemTableFactory.java | 54 ++
.../file/testutils/catalog/JsonSerdeUtil.java | 61 ++
.../testutils/catalog/TestFileSystemCatalog.java | 690 +++++++++++++++++++++
.../catalog/TestFileSystemCatalogFactory.java | 78 +++
.../org.apache.flink.table.factories.Factory | 17 +
.../testutils/TestFileSystemTableFactoryTest.java | 64 ++
.../catalog/TestFileSystemCatalogFactoryTest.java | 92 +++
.../catalog/TestFileSystemCatalogITCase.java | 89 +++
.../catalog/TestFileSystemCatalogTest.java | 379 +++++++++++
.../catalog/TestFileSystemCatalogTestBase.java | 56 ++
.../src/test/resources/log4j2-test.properties | 28 +
flink-test-utils-parent/pom.xml | 1 +
15 files changed, 1989 insertions(+)
diff --git a/flink-test-utils-parent/flink-table-filesystem-test-utils/pom.xml b/flink-test-utils-parent/flink-table-filesystem-test-utils/pom.xml
new file mode 100644
index 00000000000..e1c8fdce665
--- /dev/null
+++ b/flink-test-utils-parent/flink-table-filesystem-test-utils/pom.xml
@@ -0,0 +1,103 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ 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.
+ -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-test-utils-parent</artifactId>
+ <version>1.20-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>flink-table-filesystem-test-utils</artifactId>
+ <name>Flink : Test utils : Table Filesystem</name>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-core</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-common</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-api-java-bridge</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-files</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <!-- test dependencies -->
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-common</artifactId>
+ <version>${project.version}</version>
+ <classifier>tests</classifier>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-api-java</artifactId>
+ <version>${project.version}</version>
+ <classifier>tests</classifier>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-planner_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-planner_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ <type>test-jar</type>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-csv</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-test-utils</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
+</project>
\ No newline at end of file
diff --git a/flink-test-utils-parent/flink-table-filesystem-test-utils/src/main/java/org/apache/flink/connector/file/src/TestFileSource.java b/flink-test-utils-parent/flink-table-filesystem-test-utils/src/main/java/org/apache/flink/connector/file/src/TestFileSource.java
new file mode 100644
index 00000000000..5d0e9ace77f
--- /dev/null
+++ b/flink-test-utils-parent/flink-table-filesystem-test-utils/src/main/java/org/apache/flink/connector/file/src/TestFileSource.java
@@ -0,0 +1,189 @@
+/*
+ * 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.connector.file.src;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.api.connector.source.DynamicParallelismInference;
+import org.apache.flink.connector.file.src.assigners.FileSplitAssigner;
+import org.apache.flink.connector.file.src.assigners.LocalityAwareSplitAssigner;
+import org.apache.flink.connector.file.src.enumerate.BlockSplittingRecursiveEnumerator;
+import org.apache.flink.connector.file.src.enumerate.FileEnumerator;
+import org.apache.flink.connector.file.src.enumerate.NonSplittingRecursiveEnumerator;
+import org.apache.flink.connector.file.src.reader.BulkFormat;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.Preconditions;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.Collection;
+
+/**
+ * A unified data source that reads files - both in batch and in streaming mode. This is used only
+ * for test. Due to {@link FileSource} is a final class, so we can't extend it directly.
+ *
+ * @param <T> The type of the events/records produced by this source.
+ */
+@Internal
+public class TestFileSource<T> extends AbstractFileSource<T, FileSourceSplit>
+ implements DynamicParallelismInference {
+
+ private static final long serialVersionUID = 1L;
+
+ /** The default split assigner, a lazy locality-aware assigner. */
+ public static final FileSplitAssigner.Provider DEFAULT_SPLIT_ASSIGNER =
+ LocalityAwareSplitAssigner::new;
+
+ /**
+ * The default file enumerator used for splittable formats. The enumerator recursively
+ * enumerates files, split files that consist of multiple distributed storage blocks into
+ * multiple splits, and filters hidden files (files starting with '.' or '_'). Files with
+ * suffixes of common compression formats (for example '.gzip', '.bz2', '.xy', '.zip', ...) will
+ * not be split.
+ */
+ public static final FileEnumerator.Provider DEFAULT_SPLITTABLE_FILE_ENUMERATOR =
+ BlockSplittingRecursiveEnumerator::new;
+
+ /**
+ * The default file enumerator used for non-splittable formats. The enumerator recursively
+ * enumerates files, creates one split for the file, and filters hidden files (files starting
+ * with '.' or '_').
+ */
+ public static final FileEnumerator.Provider DEFAULT_NON_SPLITTABLE_FILE_ENUMERATOR =
+ NonSplittingRecursiveEnumerator::new;
+
+ private final boolean isStreamingMode;
+ private final ContinuousEnumerationSettings continuousEnumerationSettings;
+
+ // ------------------------------------------------------------------------
+
+ private TestFileSource(
+ final Path[] inputPaths,
+ final FileEnumerator.Provider fileEnumerator,
+ final FileSplitAssigner.Provider splitAssigner,
+ final BulkFormat<T, FileSourceSplit> readerFormat,
+ final boolean isStreamingMode,
+ @Nullable final ContinuousEnumerationSettings continuousEnumerationSettings) {
+
+ super(
+ inputPaths,
+ fileEnumerator,
+ splitAssigner,
+ readerFormat,
+ continuousEnumerationSettings);
+ this.isStreamingMode = isStreamingMode;
+ this.continuousEnumerationSettings = continuousEnumerationSettings;
+ }
+
+ @Override
+ public SimpleVersionedSerializer<FileSourceSplit> getSplitSerializer() {
+ return FileSourceSplitSerializer.INSTANCE;
+ }
+
+ @Override
+ public Boundedness getBoundedness() {
+ return isStreamingMode && continuousEnumerationSettings != null
+ ? Boundedness.CONTINUOUS_UNBOUNDED
+ : Boundedness.BOUNDED;
+ }
+
+ @Override
+ public int inferParallelism(Context dynamicParallelismContext) {
+ FileEnumerator fileEnumerator = getEnumeratorFactory().create();
+
+ Collection<FileSourceSplit> splits;
+ try {
+ splits =
+ fileEnumerator.enumerateSplits(
+ inputPaths,
+ dynamicParallelismContext.getParallelismInferenceUpperBound());
+ } catch (IOException e) {
+ throw new FlinkRuntimeException("Could not enumerate file splits", e);
+ }
+
+ return Math.min(
+ splits.size(), dynamicParallelismContext.getParallelismInferenceUpperBound());
+ }
+
+ // ------------------------------------------------------------------------
+ // Entry-point Factory Methods
+ // ------------------------------------------------------------------------
+ /**
+ * Builds a new {@code FileSource} using a {@link BulkFormat} to read batches of records from
+ * files.
+ *
+ * <p>Examples for bulk readers are compressed and vectorized formats such as ORC or Parquet.
+ */
+ public static <T> TestFileSource.TestFileSourceBuilder<T> forBulkFileFormat(
+ final BulkFormat<T, FileSourceSplit> bulkFormat, final Path... paths) {
+ Preconditions.checkNotNull(bulkFormat, "reader");
+ Preconditions.checkNotNull(paths, "paths");
+ Preconditions.checkArgument(paths.length > 0, "paths must not be empty");
+
+ return new TestFileSource.TestFileSourceBuilder<>(paths, bulkFormat);
+ }
+
+ // ------------------------------------------------------------------------
+ // Builder
+ // ------------------------------------------------------------------------
+
+ /**
+ * The builder for the {@code FileSource}, to configure the various behaviors.
+ *
+ * <p>Start building the source via one of the following methods:
+ *
+ * <ul>
+ * <li>{@link TestFileSource#forBulkFileFormat(BulkFormat, Path...)}
+ * </ul>
+ */
+ public static final class TestFileSourceBuilder<T>
+ extends AbstractFileSourceBuilder<T, FileSourceSplit, FileSource.FileSourceBuilder<T>> {
+
+ private boolean isStreamingMode = false;
+
+ TestFileSourceBuilder(Path[] inputPaths, BulkFormat<T, FileSourceSplit> readerFormat) {
+ super(
+ inputPaths,
+ readerFormat,
+ readerFormat.isSplittable()
+ ? DEFAULT_SPLITTABLE_FILE_ENUMERATOR
+ : DEFAULT_NON_SPLITTABLE_FILE_ENUMERATOR,
+ DEFAULT_SPLIT_ASSIGNER);
+ }
+
+ public TestFileSourceBuilder<T> setStreamingMode(boolean streamingMode) {
+ this.isStreamingMode = streamingMode;
+ return this;
+ }
+
+ @Override
+ public TestFileSource<T> build() {
+ return new TestFileSource<>(
+ inputPaths,
+ fileEnumerator,
+ splitAssigner,
+ readerFormat,
+ isStreamingMode,
+ continuousSourceSettings);
+ }
+ }
+}
diff --git a/flink-test-utils-parent/flink-table-filesystem-test-utils/src/main/java/org/apache/flink/connector/file/table/TestFileSystemTableSource.java b/flink-test-utils-parent/flink-table-filesystem-test-utils/src/main/java/org/apache/flink/connector/file/table/TestFileSystemTableSource.java
new file mode 100644
index 00000000000..3158f03473c
--- /dev/null
+++ b/flink-test-utils-parent/flink-table-filesystem-test-utils/src/main/java/org/apache/flink/connector/file/table/TestFileSystemTableSource.java
@@ -0,0 +1,88 @@
+/*
+ * 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.connector.file.table;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.RuntimeExecutionMode;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.configuration.ExecutionOptions;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.connector.file.src.FileSourceSplit;
+import org.apache.flink.connector.file.src.TestFileSource;
+import org.apache.flink.connector.file.src.enumerate.BlockSplittingRecursiveAllDirEnumerator;
+import org.apache.flink.connector.file.src.enumerate.NonSplittingRecursiveAllDirEnumerator;
+import org.apache.flink.connector.file.src.reader.BulkFormat;
+import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.connector.format.DecodingFormat;
+import org.apache.flink.table.connector.source.SourceProvider;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.DataType;
+
+import javax.annotation.Nullable;
+
+import java.util.List;
+
+/** Test file system table source. */
+@Internal
+public class TestFileSystemTableSource extends FileSystemTableSource {
+
+ public TestFileSystemTableSource(
+ ObjectIdentifier tableIdentifier,
+ DataType physicalRowDataType,
+ List<String> partitionKeys,
+ ReadableConfig tableOptions,
+ @Nullable DecodingFormat<BulkFormat<RowData, FileSourceSplit>> bulkReaderFormat,
+ @Nullable DecodingFormat<DeserializationSchema<RowData>> deserializationFormat) {
+ super(
+ tableIdentifier,
+ physicalRowDataType,
+ partitionKeys,
+ tableOptions,
+ bulkReaderFormat,
+ deserializationFormat);
+ }
+
+ @Override
+ protected SourceProvider createSourceProvider(BulkFormat<RowData, FileSourceSplit> bulkFormat) {
+ final TestFileSource.TestFileSourceBuilder<RowData> fileSourceBuilder =
+ TestFileSource.forBulkFileFormat(bulkFormat, paths());
+
+ tableOptions
+ .getOptional(FileSystemConnectorOptions.SOURCE_MONITOR_INTERVAL)
+ .ifPresent(fileSourceBuilder::monitorContinuously);
+ tableOptions
+ .getOptional(FileSystemConnectorOptions.SOURCE_PATH_REGEX_PATTERN)
+ .ifPresent(
+ regex ->
+ fileSourceBuilder.setFileEnumerator(
+ bulkFormat.isSplittable()
+ ? () ->
+ new BlockSplittingRecursiveAllDirEnumerator(
+ regex)
+ : () ->
+ new NonSplittingRecursiveAllDirEnumerator(
+ regex)));
+
+ boolean isStreamingMode =
+ tableOptions.get(ExecutionOptions.RUNTIME_MODE) == RuntimeExecutionMode.STREAMING;
+ fileSourceBuilder.setStreamingMode(isStreamingMode);
+
+ return SourceProvider.of(fileSourceBuilder.build());
+ }
+}
diff --git a/flink-test-utils-parent/flink-table-filesystem-test-utils/src/main/java/org/apache/flink/table/file/testutils/TestFileSystemTableFactory.java b/flink-test-utils-parent/flink-table-filesystem-test-utils/src/main/java/org/apache/flink/table/file/testutils/TestFileSystemTableFactory.java
new file mode 100644
index 00000000000..aa5cd5e17bb
--- /dev/null
+++ b/flink-test-utils-parent/flink-table-filesystem-test-utils/src/main/java/org/apache/flink/table/file/testutils/TestFileSystemTableFactory.java
@@ -0,0 +1,54 @@
+/*
+ * 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;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.connector.file.table.FileSystemTableFactory;
+import org.apache.flink.connector.file.table.TestFileSystemTableSource;
+import org.apache.flink.connector.file.table.factories.BulkReaderFormatFactory;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.factories.DeserializationFormatFactory;
+import org.apache.flink.table.factories.Factory;
+import org.apache.flink.table.factories.FactoryUtil;
+
+/** Test filesystem {@link Factory}. */
+@Internal
+public class TestFileSystemTableFactory extends FileSystemTableFactory {
+
+ public static final String IDENTIFIER = "test-filesystem";
+
+ @Override
+ public String factoryIdentifier() {
+ return IDENTIFIER;
+ }
+
+ @Override
+ public DynamicTableSource createDynamicTableSource(Context context) {
+ FactoryUtil.TableFactoryHelper helper = FactoryUtil.createTableFactoryHelper(this, context);
+ validate(helper);
+
+ return new TestFileSystemTableSource(
+ context.getObjectIdentifier(),
+ context.getPhysicalRowDataType(),
+ context.getCatalogTable().getPartitionKeys(),
+ helper.getOptions(),
+ discoverDecodingFormat(context, BulkReaderFormatFactory.class),
+ discoverDecodingFormat(context, DeserializationFormatFactory.class));
+ }
+}
diff --git a/flink-test-utils-parent/flink-table-filesystem-test-utils/src/main/java/org/apache/flink/table/file/testutils/catalog/JsonSerdeUtil.java b/flink-test-utils-parent/flink-table-filesystem-test-utils/src/main/java/org/apache/flink/table/file/testutils/catalog/JsonSerdeUtil.java
new file mode 100644
index 00000000000..755ecfd897b
--- /dev/null
+++ b/flink-test-utils-parent/flink-table-filesystem-test-utils/src/main/java/org/apache/flink/table/file/testutils/catalog/JsonSerdeUtil.java
@@ -0,0 +1,61 @@
+/*
+ * 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.util.jackson.JacksonMapperFactory;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.json.JsonReadFeature;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+
+public class JsonSerdeUtil {
+
+ /**
+ * Object mapper shared instance to serialize and deserialize the plan. Note that creating and
+ * copying of object mappers is expensive and should be avoided.
+ */
+ private static final ObjectMapper OBJECT_MAPPER_INSTANCE;
+
+ static {
+ OBJECT_MAPPER_INSTANCE =
+ JacksonMapperFactory.createObjectMapper()
+ .configure(
+ JsonReadFeature.ALLOW_UNESCAPED_CONTROL_CHARS.mappedFeature(),
+ true);
+ }
+
+ public static <T> T fromJson(String json, Class<T> clazz) {
+ try {
+ return OBJECT_MAPPER_INSTANCE.reader().readValue(json, clazz);
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ public static <T> String toJson(T t) {
+ try {
+ return OBJECT_MAPPER_INSTANCE.writerWithDefaultPrettyPrinter().writeValueAsString(t);
+ } catch (JsonProcessingException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+}
diff --git a/flink-test-utils-parent/flink-table-filesystem-test-utils/src/main/java/org/apache/flink/table/file/testutils/catalog/TestFileSystemCatalog.java b/flink-test-utils-parent/flink-table-filesystem-test-utils/src/main/java/org/apache/flink/table/file/testutils/catalog/TestFileSystemCatalog.java
new file mode 100644
index 00000000000..490dd29d608
--- /dev/null
+++ b/flink-test-utils-parent/flink-table-filesystem-test-utils/src/main/java/org/apache/flink/table/file/testutils/catalog/TestFileSystemCatalog.java
@@ -0,0 +1,690 @@
+/*
+ * 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.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;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+
+/** 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 {
+ checkArgument(
+ !StringUtils.isNullOrWhitespaceOnly(databaseName),
+ "The database name cannot be null or empty.");
+
+ return listDatabases().contains(databaseName);
+ }
+
+ @Override
+ public void createDatabase(String name, CatalogDatabase database, boolean ignoreIfExists)
+ throws DatabaseAlreadyExistException, CatalogException {
+ if (databaseExists(name)) {
+ if (ignoreIfExists) {
+ return;
+ } else {
+ throw new DatabaseAlreadyExistException(getName(), name);
+ }
+ }
+
+ if (!CollectionUtil.isNullOrEmpty(database.getProperties())) {
+ throw new CatalogException(
+ "TestFilesystem catalog doesn't support to create database with options.");
+ }
+
+ Path dbPath = new Path(catalogPath, name);
+ try {
+ fs.mkdirs(dbPath);
+ } catch (IOException e) {
+ throw new CatalogException(
+ String.format("Creating database %s occur exception.", name), e);
+ }
+ }
+
+ @Override
+ public void dropDatabase(String databaseName, boolean ignoreIfNotExists, boolean cascade)
+ throws DatabaseNotExistException, DatabaseNotEmptyException, CatalogException {
+ if (!databaseExists(databaseName)) {
+ if (ignoreIfNotExists) {
+ return;
+ } else {
+ throw new DatabaseNotExistException(getName(), databaseName);
+ }
+ }
+
+ List<String> tables = listTables(databaseName);
+ if (!tables.isEmpty() && !cascade) {
+ throw new DatabaseNotEmptyException(getName(), databaseName);
+ }
+
+ if (databaseName.equals(getDefaultDatabase())) {
+ throw new IllegalArgumentException(
+ "TestFilesystem catalog doesn't support to drop the default database.");
+ }
+
+ Path dbPath = new Path(catalogPath, databaseName);
+ try {
+ fs.delete(dbPath, true);
+ } catch (IOException e) {
+ throw new CatalogException(
+ String.format("Dropping database %s occur exception.", databaseName), e);
+ }
+ }
+
+ @Override
+ public void alterDatabase(String name, CatalogDatabase newDatabase, boolean ignoreIfNotExists)
+ throws DatabaseNotExistException, CatalogException {
+ throw new UnsupportedOperationException("alterDatabase is not implemented.");
+ }
+
+ @Override
+ public List<String> listTables(String databaseName)
+ throws DatabaseNotExistException, CatalogException {
+ if (!databaseExists(databaseName)) {
+ throw new DatabaseNotExistException(getName(), databaseName);
+ }
+
+ Path dbPath = new Path(catalogPath, databaseName);
+ try {
+ return Arrays.stream(fs.listStatus(dbPath))
+ .filter(FileStatus::isDir)
+ .map(fileStatus -> fileStatus.getPath().getName())
+ .collect(Collectors.toList());
+ } catch (IOException e) {
+ throw new CatalogException(
+ String.format("Listing table in database %s occur exception.", dbPath), e);
+ }
+ }
+
+ @Override
+ public List<String> listViews(String databaseName)
+ throws DatabaseNotExistException, CatalogException {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public CatalogBaseTable getTable(ObjectPath tablePath)
+ throws TableNotExistException, CatalogException {
+ if (!tableExists(tablePath)) {
+ throw new TableNotExistException(getName(), tablePath);
+ }
+
+ final Path tableSchemaPath =
+ tableSchemaFilePath(
+ inferTableSchemaPath(catalogPathStr, tablePath), tablePath.getObjectName());
+ final Path tableDataPath = inferTableDataPath(catalogPathStr, tablePath);
+ try {
+ FileSystemTableInfo tableInfo =
+ JsonSerdeUtil.fromJson(
+ readFileUtf8(tableSchemaPath), FileSystemTableInfo.class);
+ return deserializeTable(
+ tableInfo.getTableKind(),
+ tableInfo.getCatalogTableInfo(),
+ tableDataPath.getPath());
+ } catch (IOException e) {
+ throw new CatalogException(
+ String.format("Getting table %s occur exception.", tablePath), e);
+ }
+ }
+
+ @Override
+ public boolean tableExists(ObjectPath tablePath) throws CatalogException {
+ Path path = inferTablePath(catalogPathStr, tablePath);
+ Path tableSchemaFilePath =
+ tableSchemaFilePath(
+ inferTableSchemaPath(catalogPathStr, tablePath), tablePath.getObjectName());
+ try {
+ return fs.exists(path) && fs.exists(tableSchemaFilePath);
+ } catch (IOException e) {
+ throw new CatalogException(
+ String.format("Checking table %s exists occur exception.", tablePath), e);
+ }
+ }
+
+ @Override
+ public void dropTable(ObjectPath tablePath, boolean ignoreIfNotExists)
+ throws TableNotExistException, CatalogException {
+ if (!tableExists(tablePath)) {
+ if (ignoreIfNotExists) {
+ return;
+ } else {
+ throw new TableNotExistException(getName(), tablePath);
+ }
+ }
+
+ Path path = inferTablePath(catalogPathStr, tablePath);
+ try {
+ fs.delete(path, true);
+ } catch (IOException e) {
+ throw new CatalogException(
+ String.format("Dropping table %s occur exception.", tablePath), e);
+ }
+ }
+
+ @Override
+ public void renameTable(ObjectPath tablePath, String newTableName, boolean ignoreIfNotExists)
+ throws TableNotExistException, TableAlreadyExistException, CatalogException {
+ throw new UnsupportedOperationException("renameTable is not implemented.");
+ }
+
+ @Override
+ public void createTable(
+ ObjectPath tablePath, CatalogBaseTable catalogTable, boolean ignoreIfExists)
+ throws TableAlreadyExistException, DatabaseNotExistException, CatalogException {
+ if (!databaseExists(tablePath.getDatabaseName())) {
+ throw new DatabaseNotExistException(getName(), tablePath.getDatabaseName());
+ }
+ if (tableExists(tablePath)) {
+ if (ignoreIfExists) {
+ return;
+ } else {
+ throw new TableAlreadyExistException(getName(), tablePath);
+ }
+ }
+
+ if (catalogTable instanceof CatalogView) {
+ throw new UnsupportedOperationException(
+ "TestFilesystem catalog doesn't support to CREATE VIEW.");
+ }
+
+ Tuple4<Path, Path, Path, String> tableSchemaTuple =
+ getTableJsonInfo(tablePath, catalogTable);
+ Path path = tableSchemaTuple.f0;
+ Path tableSchemaPath = tableSchemaTuple.f1;
+ Path tableDataPath = tableSchemaTuple.f2;
+ String jsonSchema = tableSchemaTuple.f3;
+ try {
+ if (!fs.exists(path)) {
+ fs.mkdirs(path);
+ fs.mkdirs(tableSchemaPath);
+ fs.mkdirs(tableDataPath);
+ }
+
+ // write table schema
+ Path tableSchemaFilePath =
+ tableSchemaFilePath(tableSchemaPath, tablePath.getObjectName());
+ try (FSDataOutputStream os =
+ fs.create(tableSchemaFilePath, FileSystem.WriteMode.NO_OVERWRITE)) {
+ os.write(jsonSchema.getBytes(StandardCharsets.UTF_8));
+ }
+
+ } catch (IOException e) {
+ throw new CatalogException(
+ String.format("Create table %s occur exception.", tablePath), e);
+ }
+ }
+
+ @Override
+ public void alterTable(
+ ObjectPath tablePath, CatalogBaseTable newTable, boolean ignoreIfNotExists)
+ throws TableNotExistException, CatalogException {
+ throw new UnsupportedOperationException("alterTable is not implemented");
+ }
+
+ @Override
+ public void alterTable(
+ ObjectPath tablePath,
+ CatalogBaseTable newTable,
+ List<TableChange> tableChanges,
+ boolean ignoreIfNotExists)
+ throws TableNotExistException, CatalogException {
+ if (ignoreIfNotExists && !tableExists(tablePath)) {
+ return;
+ }
+
+ Tuple4<Path, Path, Path, String> tableSchemaInfo = getTableJsonInfo(tablePath, newTable);
+ Path tableSchemaPath = tableSchemaInfo.f1;
+ String jsonSchema = tableSchemaInfo.f3;
+ try {
+ if (!fs.exists(tableSchemaPath)) {
+ throw new CatalogException(
+ String.format(
+ "Table %s schema file %s doesn't exists.",
+ tablePath, tableSchemaPath));
+ }
+ // write new table schema
+ Path tableSchemaFilePath =
+ tableSchemaFilePath(tableSchemaPath, tablePath.getObjectName());
+ try (FSDataOutputStream os =
+ fs.create(tableSchemaFilePath, FileSystem.WriteMode.OVERWRITE)) {
+ os.write(jsonSchema.getBytes(StandardCharsets.UTF_8));
+ }
+
+ } catch (IOException e) {
+ throw new CatalogException(
+ String.format("Altering table %s occur exception.", tablePath), e);
+ }
+ }
+
+ @Override
+ public List<CatalogPartitionSpec> listPartitions(ObjectPath tablePath)
+ throws TableNotExistException, TableNotPartitionedException, CatalogException {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public List<CatalogPartitionSpec> listPartitions(
+ ObjectPath tablePath, CatalogPartitionSpec partitionSpec)
+ throws TableNotExistException, TableNotPartitionedException,
+ PartitionSpecInvalidException, CatalogException {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public List<CatalogPartitionSpec> listPartitionsByFilter(
+ ObjectPath tablePath, List<Expression> filters)
+ throws TableNotExistException, TableNotPartitionedException, CatalogException {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public CatalogPartition getPartition(ObjectPath tablePath, CatalogPartitionSpec partitionSpec)
+ throws PartitionNotExistException, CatalogException {
+ throw new PartitionNotExistException(getName(), tablePath, partitionSpec);
+ }
+
+ @Override
+ public boolean partitionExists(ObjectPath tablePath, CatalogPartitionSpec partitionSpec)
+ throws CatalogException {
+ return false;
+ }
+
+ @Override
+ public void createPartition(
+ ObjectPath tablePath,
+ CatalogPartitionSpec partitionSpec,
+ CatalogPartition partition,
+ boolean ignoreIfExists)
+ throws TableNotExistException, TableNotPartitionedException,
+ PartitionSpecInvalidException, PartitionAlreadyExistsException,
+ CatalogException {
+ throw new UnsupportedOperationException("createPartition is not implemented.");
+ }
+
+ @Override
+ public void dropPartition(
+ ObjectPath tablePath, CatalogPartitionSpec partitionSpec, boolean ignoreIfNotExists)
+ throws PartitionNotExistException, CatalogException {
+ throw new UnsupportedOperationException("dropPartition is not implemented.");
+ }
+
+ @Override
+ public void alterPartition(
+ ObjectPath tablePath,
+ CatalogPartitionSpec partitionSpec,
+ CatalogPartition newPartition,
+ boolean ignoreIfNotExists)
+ throws PartitionNotExistException, CatalogException {
+ throw new UnsupportedOperationException("alterPartition is not implemented.");
+ }
+
+ @Override
+ public List<String> listFunctions(String dbName)
+ throws DatabaseNotExistException, CatalogException {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public CatalogFunction getFunction(ObjectPath functionPath)
+ throws FunctionNotExistException, CatalogException {
+ throw new FunctionNotExistException(getName(), functionPath);
+ }
+
+ @Override
+ public boolean functionExists(ObjectPath functionPath) throws CatalogException {
+ return false;
+ }
+
+ @Override
+ public void createFunction(
+ ObjectPath functionPath, CatalogFunction function, boolean ignoreIfExists)
+ throws FunctionAlreadyExistException, DatabaseNotExistException, CatalogException {
+ throw new UnsupportedOperationException("createFunction is not implemented.");
+ }
+
+ @Override
+ public void alterFunction(
+ ObjectPath functionPath, CatalogFunction newFunction, boolean ignoreIfNotExists)
+ throws FunctionNotExistException, CatalogException {
+ throw new UnsupportedOperationException("alterFunction is not implemented.");
+ }
+
+ @Override
+ public void dropFunction(ObjectPath functionPath, boolean ignoreIfNotExists)
+ throws FunctionNotExistException, CatalogException {
+ throw new UnsupportedOperationException("dropFunction is not implemented.");
+ }
+
+ @Override
+ public CatalogTableStatistics getTableStatistics(ObjectPath tablePath)
+ throws TableNotExistException, CatalogException {
+ return CatalogTableStatistics.UNKNOWN;
+ }
+
+ @Override
+ public CatalogColumnStatistics getTableColumnStatistics(ObjectPath tablePath)
+ throws TableNotExistException, CatalogException {
+ return CatalogColumnStatistics.UNKNOWN;
+ }
+
+ @Override
+ public CatalogTableStatistics getPartitionStatistics(
+ ObjectPath tablePath, CatalogPartitionSpec partitionSpec)
+ throws PartitionNotExistException, CatalogException {
+ return CatalogTableStatistics.UNKNOWN;
+ }
+
+ @Override
+ public CatalogColumnStatistics getPartitionColumnStatistics(
+ ObjectPath tablePath, CatalogPartitionSpec partitionSpec)
+ throws PartitionNotExistException, CatalogException {
+ return CatalogColumnStatistics.UNKNOWN;
+ }
+
+ @Override
+ public void alterTableStatistics(
+ ObjectPath tablePath, CatalogTableStatistics tableStatistics, boolean ignoreIfNotExists)
+ throws TableNotExistException, CatalogException {
+ throw new UnsupportedOperationException("alterTableStatistics is not implemented.");
+ }
+
+ @Override
+ public void alterTableColumnStatistics(
+ ObjectPath tablePath,
+ CatalogColumnStatistics columnStatistics,
+ boolean ignoreIfNotExists)
+ throws TableNotExistException, CatalogException, TablePartitionedException {
+ throw new UnsupportedOperationException("alterTableColumnStatistics is not implemented.");
+ }
+
+ @Override
+ public void alterPartitionStatistics(
+ ObjectPath tablePath,
+ CatalogPartitionSpec partitionSpec,
+ CatalogTableStatistics partitionStatistics,
+ boolean ignoreIfNotExists)
+ throws PartitionNotExistException, CatalogException {
+ throw new UnsupportedOperationException("alterPartitionStatistics is not implemented.");
+ }
+
+ @Override
+ public void alterPartitionColumnStatistics(
+ ObjectPath tablePath,
+ CatalogPartitionSpec partitionSpec,
+ CatalogColumnStatistics columnStatistics,
+ boolean ignoreIfNotExists)
+ throws PartitionNotExistException, CatalogException {
+ throw new UnsupportedOperationException(
+ "alterPartitionColumnStatistics is not implemented.");
+ }
+
+ private Tuple4<Path, Path, Path, String> getTableJsonInfo(
+ ObjectPath tablePath, CatalogBaseTable catalogTable) {
+ final Path path = inferTablePath(catalogPathStr, tablePath);
+ final Path tableSchemaPath = inferTableSchemaPath(catalogPathStr, tablePath);
+ final Path tableDataPathStr = inferTableDataPath(catalogPathStr, tablePath);
+
+ ResolvedCatalogBaseTable<?> resolvedCatalogBaseTable =
+ (ResolvedCatalogBaseTable<?>) catalogTable;
+ Map<String, String> catalogTableInfo = serializeTable(resolvedCatalogBaseTable);
+ CatalogBaseTable.TableKind tableKind = catalogTable.getTableKind();
+ FileSystemTableInfo tableInfo = new FileSystemTableInfo(tableKind, catalogTableInfo);
+
+ String jsonSchema = JsonSerdeUtil.toJson(tableInfo);
+ return Tuple4.of(path, tableSchemaPath, tableDataPathStr, jsonSchema);
+ }
+
+ /** Read file to UTF_8 decoding. */
+ private String readFileUtf8(Path path) throws IOException {
+ try (FSDataInputStream in = path.getFileSystem().open(path)) {
+ BufferedReader reader =
+ new BufferedReader(new InputStreamReader(in, StandardCharsets.UTF_8));
+ StringBuilder builder = new StringBuilder();
+ String line;
+ while ((line = reader.readLine()) != null) {
+ builder.append(line);
+ }
+ return builder.toString();
+ }
+ }
+
+ private Path inferTablePath(String catalogPath, ObjectPath tablePath) {
+ return new Path(
+ String.format(
+ "%s/%s/%s",
+ catalogPath, tablePath.getDatabaseName(), tablePath.getObjectName()));
+ }
+
+ private Path inferTableDataPath(String catalogPath, ObjectPath tablePath) {
+ return new Path(
+ String.format(
+ "%s/%s/%s/%s",
+ catalogPath,
+ tablePath.getDatabaseName(),
+ tablePath.getObjectName(),
+ DATA_PATH));
+ }
+
+ private Path inferTableSchemaPath(String catalogPath, ObjectPath tablePath) {
+ return new Path(
+ String.format(
+ "%s/%s/%s/%s",
+ catalogPath,
+ tablePath.getDatabaseName(),
+ tablePath.getObjectName(),
+ SCHEMA_PATH));
+ }
+
+ private Path tableSchemaFilePath(Path tableSchemaPath, String tableName) {
+ return new Path(tableSchemaPath, tableName + "_schema" + SCHEMA_FILE_EXTENSION);
+ }
+
+ private Map<String, String> serializeTable(
+ ResolvedCatalogBaseTable<?> resolvedCatalogBaseTable) {
+ if (resolvedCatalogBaseTable instanceof ResolvedCatalogTable) {
+ return CatalogPropertiesUtil.serializeCatalogTable(
+ (ResolvedCatalogTable) resolvedCatalogBaseTable);
+ } else if (resolvedCatalogBaseTable instanceof ResolvedCatalogMaterializedTable) {
+ return CatalogPropertiesUtil.serializeCatalogMaterializedTable(
+ (ResolvedCatalogMaterializedTable) resolvedCatalogBaseTable);
+ }
+
+ throw new IllegalArgumentException(
+ "Unknown kind of resolved catalog base table: "
+ + resolvedCatalogBaseTable.getClass());
+ }
+
+ private CatalogBaseTable deserializeTable(
+ CatalogBaseTable.TableKind tableKind,
+ Map<String, String> properties,
+ String tableDataPath) {
+ if (CatalogBaseTable.TableKind.TABLE == tableKind) {
+ CatalogTable catalogTable = CatalogPropertiesUtil.deserializeCatalogTable(properties);
+ // put table data path
+ Map<String, String> options = new HashMap<>(catalogTable.getOptions());
+ options.put(FileSystemConnectorOptions.PATH.key(), tableDataPath);
+ return catalogTable.copy(options);
+ } else if (CatalogBaseTable.TableKind.MATERIALIZED_TABLE == tableKind) {
+ CatalogMaterializedTable catalogMaterializedTable =
+ CatalogPropertiesUtil.deserializeCatalogMaterializedTable(properties);
+ // put table data path
+ Map<String, String> options = new HashMap<>(catalogMaterializedTable.getOptions());
+ options.put(FileSystemConnectorOptions.PATH.key(), tableDataPath);
+ return catalogMaterializedTable.copy(options);
+ }
+
+ throw new IllegalArgumentException("Unknown catalog base table kind: " + tableKind);
+ }
+
+ /** The pojo class represents serializable catalog base table info. */
+ public static class FileSystemTableInfo {
+
+ private final CatalogBaseTable.TableKind tableKind;
+ private final Map<String, String> catalogTableInfo;
+
+ @JsonCreator
+ public FileSystemTableInfo(
+ @JsonProperty("tableKind") CatalogBaseTable.TableKind tableKind,
+ @JsonProperty("catalogTableInfo") Map<String, String> catalogTableInfo) {
+ this.tableKind = tableKind;
+ this.catalogTableInfo = catalogTableInfo;
+ }
+
+ public CatalogBaseTable.TableKind getTableKind() {
+ return tableKind;
+ }
+
+ public Map<String, String> getCatalogTableInfo() {
+ return catalogTableInfo;
+ }
+ }
+}
diff --git a/flink-test-utils-parent/flink-table-filesystem-test-utils/src/main/java/org/apache/flink/table/file/testutils/catalog/TestFileSystemCatalogFactory.java b/flink-test-utils-parent/flink-table-filesystem-test-utils/src/main/java/org/apache/flink/table/file/testutils/catalog/TestFileSystemCatalogFactory.java
new file mode 100644
index 00000000000..41999468d52
--- /dev/null
+++ b/flink-test-utils-parent/flink-table-filesystem-test-utils/src/main/java/org/apache/flink/table/file/testutils/catalog/TestFileSystemCatalogFactory.java
@@ -0,0 +1,78 @@
+/*
+ * 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.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+import org.apache.flink.table.catalog.Catalog;
+import org.apache.flink.table.catalog.CommonCatalogOptions;
+import org.apache.flink.table.factories.CatalogFactory;
+import org.apache.flink.table.factories.FactoryUtil;
+
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+
+import static org.apache.flink.table.file.testutils.TestFileSystemTableFactory.IDENTIFIER;
+
+/** Factory for {@link TestFileSystemCatalog}. */
+public class TestFileSystemCatalogFactory implements CatalogFactory {
+
+ public static final ConfigOption<String> PATH =
+ ConfigOptions.key("path")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Catalog base DFS path, used for inferring the sink table path. "
+ + "The default strategy for a table path is: ${catalog.path}/${db_name}/${table_name}");
+
+ public static final ConfigOption<String> DEFAULT_DATABASE =
+ ConfigOptions.key(CommonCatalogOptions.DEFAULT_DATABASE_KEY)
+ .stringType()
+ .defaultValue("default");
+
+ @Override
+ public String factoryIdentifier() {
+ return IDENTIFIER;
+ }
+
+ @Override
+ public Catalog createCatalog(Context context) {
+ final FactoryUtil.CatalogFactoryHelper helper =
+ FactoryUtil.createCatalogFactoryHelper(this, context);
+ helper.validate();
+ return new TestFileSystemCatalog(
+ helper.getOptions().get(PATH),
+ context.getName(),
+ helper.getOptions().get(DEFAULT_DATABASE));
+ }
+
+ @Override
+ public Set<ConfigOption<?>> requiredOptions() {
+ Set<ConfigOption<?>> set = new HashSet<>();
+ set.add(PATH);
+ set.add(DEFAULT_DATABASE);
+ return set;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> optionalOptions() {
+ return Collections.emptySet();
+ }
+}
diff --git a/flink-test-utils-parent/flink-table-filesystem-test-utils/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory b/flink-test-utils-parent/flink-table-filesystem-test-utils/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
new file mode 100644
index 00000000000..7977248c5be
--- /dev/null
+++ b/flink-test-utils-parent/flink-table-filesystem-test-utils/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
@@ -0,0 +1,17 @@
+# 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.
+
+org.apache.flink.table.file.testutils.catalog.TestFileSystemCatalogFactory
+org.apache.flink.table.file.testutils.TestFileSystemTableFactory
diff --git a/flink-test-utils-parent/flink-table-filesystem-test-utils/src/test/java/org/apache/flink/table/file/testutils/TestFileSystemTableFactoryTest.java b/flink-test-utils-parent/flink-table-filesystem-test-utils/src/test/java/org/apache/flink/table/file/testutils/TestFileSystemTableFactoryTest.java
new file mode 100644
index 00000000000..b02a7f124ad
--- /dev/null
+++ b/flink-test-utils-parent/flink-table-filesystem-test-utils/src/test/java/org/apache/flink/table/file/testutils/TestFileSystemTableFactoryTest.java
@@ -0,0 +1,64 @@
+/*
+ * 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;
+
+import org.apache.flink.connector.file.table.FileSystemTableSink;
+import org.apache.flink.connector.file.table.TestFileSystemTableSource;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.factories.FactoryUtil;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.apache.flink.table.factories.utils.FactoryMocks.createTableSink;
+import static org.apache.flink.table.factories.utils.FactoryMocks.createTableSource;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link TestFileSystemTableFactory}. */
+public class TestFileSystemTableFactoryTest {
+
+ private static final ResolvedSchema SCHEMA =
+ ResolvedSchema.of(
+ Column.physical("f0", DataTypes.STRING()),
+ Column.physical("f1", DataTypes.BIGINT()),
+ Column.physical("f2", DataTypes.BIGINT()));
+
+ @Test
+ void testCreateSourceSink() {
+ Map<String, String> options = new HashMap<>();
+ options.put(FactoryUtil.CONNECTOR.key(), "test-filesystem");
+ options.put("path", "/tmp");
+ options.put("format", "testcsv");
+
+ // test ignore format options
+ options.put("testcsv.my_option", "my_value");
+
+ DynamicTableSource source = createTableSource(SCHEMA, options);
+ assertThat(source).isInstanceOf(TestFileSystemTableSource.class);
+
+ DynamicTableSink sink = createTableSink(SCHEMA, options);
+ assertThat(sink).isInstanceOf(FileSystemTableSink.class);
+ }
+}
diff --git a/flink-test-utils-parent/flink-table-filesystem-test-utils/src/test/java/org/apache/flink/table/file/testutils/catalog/TestFileSystemCatalogFactoryTest.java b/flink-test-utils-parent/flink-table-filesystem-test-utils/src/test/java/org/apache/flink/table/file/testutils/catalog/TestFileSystemCatalogFactoryTest.java
new file mode 100644
index 00000000000..4b77a979f93
--- /dev/null
+++ b/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_DEFAULT_DATABASE);
+ 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_DEFAULT_DATABASE);
+ }
+}
diff --git a/flink-test-utils-parent/flink-table-filesystem-test-utils/src/test/java/org/apache/flink/table/file/testutils/catalog/TestFileSystemCatalogITCase.java b/flink-test-utils-parent/flink-table-filesystem-test-utils/src/test/java/org/apache/flink/table/file/testutils/catalog/TestFileSystemCatalogITCase.java
new file mode 100644
index 00000000000..e418f6c80a1
--- /dev/null
+++ b/flink-test-utils-parent/flink-table-filesystem-test-utils/src/test/java/org/apache/flink/table/file/testutils/catalog/TestFileSystemCatalogITCase.java
@@ -0,0 +1,89 @@
+/*
+ * 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.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CloseableIterator;
+
+import org.apache.flink.shaded.guava31.com.google.common.collect.Lists;
+
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Arrays;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Integration tests for {@link TestFileSystemCatalog}. */
+public class TestFileSystemCatalogITCase extends TestFileSystemCatalogTestBase {
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testReadAndWriteTestFileSystemTable(boolean isStreamingMode) throws Exception {
+ TableEnvironment tEnv =
+ TableEnvironment.create(
+ isStreamingMode
+ ? EnvironmentSettings.inStreamingMode()
+ : EnvironmentSettings.inBatchMode());
+
+ tEnv.registerCatalog(TEST_CATALOG, catalog);
+ tEnv.useCatalog(TEST_CATALOG);
+
+ tEnv.executeSql(
+ "CREATE TABLE CsvTable (\n"
+ + " id BIGINT,\n"
+ + " user_name STRING,\n"
+ + " message STRING,\n"
+ + " log_ts STRING\n"
+ + ") WITH (\n"
+ + " 'format' = 'csv'\n"
+ + ")");
+
+ tEnv.getConfig().getConfiguration().setString("parallelism.default", "1");
+ tEnv.executeSql(
+ String.format(
+ "INSERT INTO %s.%s.CsvTable VALUES\n"
+ + "(1001, 'user1', 'hello world', '2021-06-10 10:00:00'),\n"
+ + "(1002, 'user2', 'hi', '2021-06-10 10:01:00'),\n"
+ + "(1003, 'user3', 'ciao', '2021-06-10 10:02:00'),\n"
+ + "(1004, 'user4', '你好', '2021-06-10 10:03:00')",
+ TEST_CATALOG, TEST_DEFAULT_DATABASE))
+ .await();
+
+ CloseableIterator<Row> result =
+ tEnv.executeSql(
+ String.format(
+ "SELECT * FROM %s.%s.CsvTable",
+ TEST_CATALOG, TEST_DEFAULT_DATABASE))
+ .collect();
+
+ // assert query result
+ List<Row> expected =
+ Arrays.asList(
+ Row.of(1001L, "user1", "hello world", "2021-06-10 10:00:00"),
+ Row.of(1002L, "user2", "hi", "2021-06-10 10:01:00"),
+ Row.of(1003L, "user3", "ciao", "2021-06-10 10:02:00"),
+ Row.of(1004L, "user4", "你好", "2021-06-10 10:03:00"));
+
+ assertThat(expected).isEqualTo((Lists.newArrayList(result)));
+ }
+}
diff --git a/flink-test-utils-parent/flink-table-filesystem-test-utils/src/test/java/org/apache/flink/table/file/testutils/catalog/TestFileSystemCatalogTest.java b/flink-test-utils-parent/flink-table-filesystem-test-utils/src/test/java/org/apache/flink/table/file/testutils/catalog/TestFileSystemCatalogTest.java
new file mode 100644
index 00000000000..07b11583d4f
--- /dev/null
+++ b/flink-test-utils-parent/flink-table-filesystem-test-utils/src/test/java/org/apache/flink/table/file/testutils/catalog/TestFileSystemCatalogTest.java
@@ -0,0 +1,379 @@
+/*
+ * 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.table.api.DataTypes;
+import org.apache.flink.table.api.Schema;
+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.CatalogMaterializedTable;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.ResolvedCatalogMaterializedTable;
+import org.apache.flink.table.catalog.ResolvedCatalogTable;
+import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.catalog.TestSchemaResolver;
+import org.apache.flink.table.catalog.UniqueConstraint;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
+import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
+import org.apache.flink.table.catalog.exceptions.TableNotExistException;
+import org.apache.flink.table.refresh.RefreshHandler;
+
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+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 static org.apache.flink.connector.file.table.FileSystemConnectorOptions.PATH;
+import static org.apache.flink.table.file.testutils.catalog.TestFileSystemCatalog.DATA_PATH;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+/** Test for {@link TestFileSystemCatalog}. */
+public class TestFileSystemCatalogTest extends TestFileSystemCatalogTestBase {
+
+ private static final List<Column> CREATE_COLUMNS =
+ Arrays.asList(
+ Column.physical("id", DataTypes.BIGINT()),
+ Column.physical("name", DataTypes.VARCHAR(20)),
+ Column.physical("age", DataTypes.INT()),
+ Column.physical("tss", DataTypes.TIMESTAMP(3)),
+ Column.physical("partition", DataTypes.VARCHAR(10)));
+ private static final UniqueConstraint CONSTRAINTS =
+ UniqueConstraint.primaryKey("primary_constraint", Collections.singletonList("id"));
+ private static final List<String> PARTITION_KEYS = Collections.singletonList("partition");
+
+ private static final ResolvedSchema CREATE_RESOLVED_SCHEMA =
+ new ResolvedSchema(CREATE_COLUMNS, Collections.emptyList(), CONSTRAINTS);
+
+ private static final Schema CREATE_SCHEMA =
+ Schema.newBuilder().fromResolvedSchema(CREATE_RESOLVED_SCHEMA).build();
+
+ private static final Map<String, String> EXPECTED_OPTIONS = new HashMap<>();
+
+ static {
+ EXPECTED_OPTIONS.put("source.monitor-interval", "5S");
+ EXPECTED_OPTIONS.put("auto-compaction", "true");
+ }
+
+ private static final ResolvedCatalogTable EXPECTED_CATALOG_TABLE =
+ new ResolvedCatalogTable(
+ CatalogTable.newBuilder()
+ .schema(CREATE_SCHEMA)
+ .comment("test table")
+ .partitionKeys(PARTITION_KEYS)
+ .options(EXPECTED_OPTIONS)
+ .build(),
+ CREATE_RESOLVED_SCHEMA);
+
+ private static final String DEFINITION_QUERY = "SELECT id, region, county FROM T";
+ private static final Duration FRESHNESS = Duration.ofMinutes(3);
+ private static final ResolvedCatalogMaterializedTable EXPECTED_CATALOG_MATERIALIZED_TABLE =
+ new ResolvedCatalogMaterializedTable(
+ CatalogMaterializedTable.newBuilder()
+ .schema(CREATE_SCHEMA)
+ .comment("test materialized table")
+ .partitionKeys(PARTITION_KEYS)
+ .options(EXPECTED_OPTIONS)
+ .definitionQuery(DEFINITION_QUERY)
+ .freshness(FRESHNESS)
+ .logicalRefreshMode(
+ CatalogMaterializedTable.LogicalRefreshMode.AUTOMATIC)
+ .refreshMode(CatalogMaterializedTable.RefreshMode.CONTINUOUS)
+ .refreshStatus(CatalogMaterializedTable.RefreshStatus.INITIALIZING)
+ .build(),
+ CREATE_RESOLVED_SCHEMA);
+
+ private static final TestRefreshHandler REFRESH_HANDLER =
+ new TestRefreshHandler("jobID: xxx, clusterId: yyy");
+
+ @Test
+ public void testListDatabases() {
+ List<String> actual = catalog.listDatabases();
+ assertThat(actual.contains(TEST_DEFAULT_DATABASE)).isTrue();
+ assertThat(actual.contains(NONE_EXIST_DATABASE)).isFalse();
+ }
+
+ @Test
+ public void testDatabaseExists() {
+ assertThat(catalog.databaseExists(TEST_DEFAULT_DATABASE)).isTrue();
+ assertThat(catalog.databaseExists(NONE_EXIST_DATABASE)).isFalse();
+ }
+
+ @Test
+ public void testCreateAndDropDatabase() throws Exception {
+ CatalogDatabase expected = new CatalogDatabaseImpl(Collections.emptyMap(), null);
+ catalog.createDatabase("db1", expected, true);
+
+ CatalogDatabase actual = catalog.getDatabase("db1");
+ assertThat(catalog.listDatabases().contains("db1")).isTrue();
+ assertThat(actual.getProperties()).isEqualTo(expected.getProperties());
+
+ // create exist database
+ assertThrows(
+ DatabaseAlreadyExistException.class,
+ () -> catalog.createDatabase("db1", expected, false));
+
+ // drop exist database
+ catalog.dropDatabase("db1", true);
+ assertThat(catalog.listDatabases().contains("db1")).isFalse();
+
+ // drop non-exist database
+ assertThrows(
+ DatabaseNotExistException.class,
+ () -> catalog.dropDatabase(NONE_EXIST_DATABASE, false));
+ }
+
+ @Test
+ public void testCreateDatabaseWithOptions() {
+ Map<String, String> options = new HashMap<>();
+ options.put("k1", "v1");
+ options.put("k2", "v2");
+
+ assertThrows(
+ CatalogException.class,
+ () -> catalog.createDatabase("db1", new CatalogDatabaseImpl(options, null), true));
+ }
+
+ @Test
+ public void testCreateAndGetCatalogTable() throws Exception {
+ ObjectPath tablePath = new ObjectPath(TEST_DEFAULT_DATABASE, "tb1");
+ // test create table
+ catalog.createTable(tablePath, EXPECTED_CATALOG_TABLE, true);
+
+ // test table exist
+ assertThat(catalog.tableExists(tablePath)).isTrue();
+
+ Map<String, String> expectedOptions = new HashMap<>(EXPECTED_OPTIONS);
+ expectedOptions.put(
+ PATH.key(),
+ String.format(
+ "%s/%s/%s/%s",
+ tempFile.getAbsolutePath(),
+ tablePath.getDatabaseName(),
+ tablePath.getObjectName(),
+ DATA_PATH));
+
+ // test get table
+ CatalogBaseTable actualTable = catalog.getTable(tablePath);
+ // validate table type
+ assertThat(actualTable.getTableKind()).isEqualTo(CatalogBaseTable.TableKind.TABLE);
+
+ CatalogTable actualCatalogTable = (CatalogTable) actualTable;
+ // validate schema
+ assertThat(actualCatalogTable.getUnresolvedSchema().resolve(new TestSchemaResolver()))
+ .isEqualTo(CREATE_RESOLVED_SCHEMA);
+ // validate partition key
+ assertThat(actualCatalogTable.getPartitionKeys()).isEqualTo(PARTITION_KEYS);
+ // validate options
+ assertThat(actualCatalogTable.getOptions()).isEqualTo(expectedOptions);
+
+ // test create exist table
+ assertThrows(
+ TableAlreadyExistException.class,
+ () -> catalog.createTable(tablePath, EXPECTED_CATALOG_TABLE, false));
+ }
+
+ @Test
+ public void testCreateAndGetCatalogMaterializedTable() throws Exception {
+ ObjectPath tablePath = new ObjectPath(TEST_DEFAULT_DATABASE, "tb2");
+ // test create materialized table
+ catalog.createTable(tablePath, EXPECTED_CATALOG_MATERIALIZED_TABLE, true);
+
+ // test materialized table exist
+ assertThat(catalog.tableExists(tablePath)).isTrue();
+
+ Map<String, String> expectedOptions = new HashMap<>(EXPECTED_OPTIONS);
+ expectedOptions.put(
+ PATH.key(),
+ String.format(
+ "%s/%s/%s/%s",
+ tempFile.getAbsolutePath(),
+ tablePath.getDatabaseName(),
+ tablePath.getObjectName(),
+ DATA_PATH));
+
+ // test get materialized table
+ CatalogBaseTable actualTable = catalog.getTable(tablePath);
+ // validate table type
+ assertThat(actualTable.getTableKind())
+ .isEqualTo(CatalogBaseTable.TableKind.MATERIALIZED_TABLE);
+
+ CatalogMaterializedTable actualMaterializedTable = (CatalogMaterializedTable) actualTable;
+ // validate schema
+ assertThat(actualMaterializedTable.getUnresolvedSchema().resolve(new TestSchemaResolver()))
+ .isEqualTo(CREATE_RESOLVED_SCHEMA);
+ // validate partition key
+ assertThat(actualMaterializedTable.getPartitionKeys()).isEqualTo(PARTITION_KEYS);
+ // validate options
+ assertThat(actualMaterializedTable.getOptions()).isEqualTo(expectedOptions);
+ // validate definition query
+ assertThat(actualMaterializedTable.getDefinitionQuery()).isEqualTo(DEFINITION_QUERY);
+ // validate freshness
+ assertThat(actualMaterializedTable.getFreshness()).isEqualTo(FRESHNESS);
+ // validate logical refresh mode
+ assertThat(actualMaterializedTable.getLogicalRefreshMode())
+ .isEqualTo(CatalogMaterializedTable.LogicalRefreshMode.AUTOMATIC);
+ // validate refresh mode
+ assertThat(actualMaterializedTable.getRefreshMode())
+ .isEqualTo(CatalogMaterializedTable.RefreshMode.CONTINUOUS);
+ // validate refresh status
+ assertThat(actualMaterializedTable.getRefreshStatus())
+ .isEqualTo(CatalogMaterializedTable.RefreshStatus.INITIALIZING);
+ // validate refresh handler
+ assertThat(actualMaterializedTable.getRefreshHandlerDescription())
+ .isEqualTo(Optional.empty());
+ assertThat(actualMaterializedTable.getSerializedRefreshHandler()).isNull();
+
+ // test create exist materialized table
+ assertThrows(
+ TableAlreadyExistException.class,
+ () -> catalog.createTable(tablePath, EXPECTED_CATALOG_MATERIALIZED_TABLE, false));
+ }
+
+ @Test
+ public void testListTable() throws Exception {
+ ObjectPath tablePath1 = new ObjectPath(TEST_DEFAULT_DATABASE, "tb1");
+ ObjectPath tablePath2 = new ObjectPath(TEST_DEFAULT_DATABASE, "tb2");
+
+ // create table
+ catalog.createTable(tablePath1, EXPECTED_CATALOG_TABLE, true);
+ catalog.createTable(tablePath2, EXPECTED_CATALOG_MATERIALIZED_TABLE, true);
+
+ // test list table
+ List<String> tables = catalog.listTables(TEST_DEFAULT_DATABASE);
+ assertThat(tables.contains(tablePath1.getObjectName())).isTrue();
+ assertThat(tables.contains(tablePath2.getObjectName())).isTrue();
+
+ // test list non-exist database table
+ assertThrows(
+ DatabaseNotExistException.class, () -> catalog.listTables(NONE_EXIST_DATABASE));
+ }
+
+ @Test
+ public void testAlterCatalogTable() throws Exception {
+ ObjectPath tablePath = new ObjectPath(TEST_DEFAULT_DATABASE, "tb1");
+ // test create table
+ catalog.createTable(tablePath, EXPECTED_CATALOG_TABLE, true);
+
+ // test table exist
+ assertThat(catalog.tableExists(tablePath)).isTrue();
+
+ // alter table options
+ Map<String, String> options = new HashMap<>();
+ options.put("auto-compaction", "true");
+ options.put("sink.parallelism", "5");
+ ResolvedCatalogTable updatedResolvedTable = EXPECTED_CATALOG_TABLE.copy(options);
+ catalog.alterTable(tablePath, updatedResolvedTable, Collections.emptyList(), false);
+
+ // test get table
+ CatalogBaseTable actualTable = catalog.getTable(tablePath);
+ // validate table type
+ assertThat(actualTable.getTableKind()).isEqualTo(CatalogBaseTable.TableKind.TABLE);
+
+ // validate options
+ Map<String, String> expectedOptions = new HashMap<>(options);
+ expectedOptions.put(
+ PATH.key(),
+ String.format(
+ "%s/%s/%s/%s",
+ tempFile.getAbsolutePath(),
+ tablePath.getDatabaseName(),
+ tablePath.getObjectName(),
+ DATA_PATH));
+ assertThat(actualTable.getOptions()).isEqualTo(expectedOptions);
+ }
+
+ @Test
+ public void testAlterCatalogMaterializedTable() throws Exception {
+ ObjectPath tablePath = new ObjectPath(TEST_DEFAULT_DATABASE, "tb2");
+ // test create materialized table
+ catalog.createTable(tablePath, EXPECTED_CATALOG_MATERIALIZED_TABLE, true);
+
+ // test materialized table exist
+ assertThat(catalog.tableExists(tablePath)).isTrue();
+
+ // alter materialized table refresh handler
+ ResolvedCatalogMaterializedTable updatedMaterializedTable =
+ EXPECTED_CATALOG_MATERIALIZED_TABLE.copy(
+ CatalogMaterializedTable.RefreshStatus.ACTIVATED,
+ REFRESH_HANDLER.asSummaryString(),
+ REFRESH_HANDLER.toBytes());
+ catalog.alterTable(tablePath, updatedMaterializedTable, Collections.emptyList(), false);
+
+ // test get materialized table
+ CatalogBaseTable actualTable = catalog.getTable(tablePath);
+ // validate table type
+ assertThat(actualTable.getTableKind())
+ .isEqualTo(CatalogBaseTable.TableKind.MATERIALIZED_TABLE);
+
+ CatalogMaterializedTable actualMaterializedTable = (CatalogMaterializedTable) actualTable;
+ // validate refresh status
+ assertThat(actualMaterializedTable.getRefreshStatus())
+ .isEqualTo(CatalogMaterializedTable.RefreshStatus.ACTIVATED);
+ // validate refresh handler
+ assertThat(actualMaterializedTable.getRefreshHandlerDescription().get())
+ .isEqualTo(REFRESH_HANDLER.asSummaryString());
+ assertThat(actualMaterializedTable.getSerializedRefreshHandler())
+ .isEqualTo(REFRESH_HANDLER.toBytes());
+ }
+
+ @Test
+ public void testDropTable() throws Exception {
+ ObjectPath tablePath = new ObjectPath(TEST_DEFAULT_DATABASE, "tb1");
+ // create table
+ catalog.createTable(tablePath, EXPECTED_CATALOG_TABLE, true);
+
+ // test drop table
+ catalog.dropTable(tablePath, true);
+ assertThat(catalog.tableExists(tablePath)).isFalse();
+
+ // drop non-exist table
+ assertThrows(
+ TableNotExistException.class,
+ () -> catalog.dropTable(new ObjectPath(TEST_DEFAULT_DATABASE, "non_exist"), false));
+ }
+
+ private static class TestRefreshHandler implements RefreshHandler {
+
+ private final String handlerString;
+
+ public TestRefreshHandler(String handlerString) {
+ this.handlerString = handlerString;
+ }
+
+ @Override
+ public String asSummaryString() {
+ return "test refresh handler";
+ }
+
+ public byte[] toBytes() {
+ return handlerString.getBytes();
+ }
+ }
+}
diff --git a/flink-test-utils-parent/flink-table-filesystem-test-utils/src/test/java/org/apache/flink/table/file/testutils/catalog/TestFileSystemCatalogTestBase.java b/flink-test-utils-parent/flink-table-filesystem-test-utils/src/test/java/org/apache/flink/table/file/testutils/catalog/TestFileSystemCatalogTestBase.java
new file mode 100644
index 00000000000..8cd5fc306e6
--- /dev/null
+++ b/flink-test-utils-parent/flink-table-filesystem-test-utils/src/test/java/org/apache/flink/table/file/testutils/catalog/TestFileSystemCatalogTestBase.java
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.file.testutils.catalog;
+
+import org.apache.flink.test.util.AbstractTestBase;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+
+/** Base class for test filesystem catalog. */
+public abstract class TestFileSystemCatalogTestBase extends AbstractTestBase {
+
+ protected static final String TEST_CATALOG = "test_catalog";
+ protected static final String TEST_DEFAULT_DATABASE = "test_db";
+ protected static final String NONE_EXIST_DATABASE = "none_exist_database";
+
+ protected TestFileSystemCatalog catalog;
+
+ @TempDir File tempFile;
+
+ @BeforeEach
+ void before() {
+ File testDb = new File(tempFile, TEST_DEFAULT_DATABASE);
+ testDb.mkdir();
+
+ String catalogPathStr = tempFile.getAbsolutePath();
+ catalog = new TestFileSystemCatalog(catalogPathStr, TEST_CATALOG, TEST_DEFAULT_DATABASE);
+ catalog.open();
+ }
+
+ @AfterEach
+ void close() {
+ if (catalog != null) {
+ catalog.close();
+ }
+ }
+}
diff --git a/flink-test-utils-parent/flink-table-filesystem-test-utils/src/test/resources/log4j2-test.properties b/flink-test-utils-parent/flink-table-filesystem-test-utils/src/test/resources/log4j2-test.properties
new file mode 100644
index 00000000000..8bb9fe60fd0
--- /dev/null
+++ b/flink-test-utils-parent/flink-table-filesystem-test-utils/src/test/resources/log4j2-test.properties
@@ -0,0 +1,28 @@
+#
+# 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.
+#
+
+# Set root logger level to OFF to not flood build logs
+# set manually to INFO for debugging purposes
+rootLogger.level=OFF
+rootLogger.appenderRef.test.ref = TestLogger
+
+appender.testlogger.name = TestLogger
+appender.testlogger.type = CONSOLE
+appender.testlogger.target = SYSTEM_ERR
+appender.testlogger.layout.type = PatternLayout
+appender.testlogger.layout.pattern = %-4r [%t] %-5p %c %x - %m%n
diff --git a/flink-test-utils-parent/pom.xml b/flink-test-utils-parent/pom.xml
index 0c6a7dc08ce..96273501e20 100644
--- a/flink-test-utils-parent/pom.xml
+++ b/flink-test-utils-parent/pom.xml
@@ -40,6 +40,7 @@ under the License.
<module>flink-connector-test-utils</module>
<module>flink-clients-test-utils</module>
<module>flink-migration-test-utils</module>
+ <module>flink-table-filesystem-test-utils</module>
</modules>
</project>
(flink) 01/03: [FLINK-35189][table-common] CatalogPropertiesUtil support serialize and deserialize materialized table
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ron pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 00d69e53fe7b263293608f0a949525441b82e09e
Author: fengli <ld...@163.com>
AuthorDate: Wed Apr 24 18:11:03 2024 +0800
[FLINK-35189][table-common] CatalogPropertiesUtil support serialize and deserialize materialized table
---
.../flink/table/catalog/CatalogPropertiesUtil.java | 140 ++++++++++++++++++++-
1 file changed, 138 insertions(+), 2 deletions(-)
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/CatalogPropertiesUtil.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/CatalogPropertiesUtil.java
index 909e8f1c562..9bb6280cc7d 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/CatalogPropertiesUtil.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/catalog/CatalogPropertiesUtil.java
@@ -28,9 +28,12 @@ import org.apache.flink.table.catalog.exceptions.CatalogException;
import org.apache.flink.table.expressions.ResolvedExpression;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.util.StringUtils;
import javax.annotation.Nullable;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -117,6 +120,56 @@ public final class CatalogPropertiesUtil {
}
}
+ /**
+ * Serializes the given {@link ResolvedCatalogMaterializedTable} into a map of string
+ * properties.
+ */
+ public static Map<String, String> serializeCatalogMaterializedTable(
+ ResolvedCatalogMaterializedTable resolvedMaterializedTable) {
+ try {
+ final Map<String, String> properties = new HashMap<>();
+
+ serializeResolvedSchema(properties, resolvedMaterializedTable.getResolvedSchema());
+
+ final String comment = resolvedMaterializedTable.getComment();
+ if (comment != null && comment.length() > 0) {
+ properties.put(COMMENT, comment);
+ }
+
+ final Optional<Long> snapshot = resolvedMaterializedTable.getSnapshot();
+ snapshot.ifPresent(snapshotId -> properties.put(SNAPSHOT, Long.toString(snapshotId)));
+
+ serializePartitionKeys(properties, resolvedMaterializedTable.getPartitionKeys());
+
+ properties.putAll(resolvedMaterializedTable.getOptions());
+
+ properties.put(DEFINITION_QUERY, resolvedMaterializedTable.getDefinitionQuery());
+ properties.put(FRESHNESS, resolvedMaterializedTable.getFreshness().toString());
+
+ properties.put(
+ LOGICAL_REFRESH_MODE, resolvedMaterializedTable.getLogicalRefreshMode().name());
+ properties.put(REFRESH_MODE, resolvedMaterializedTable.getRefreshMode().name());
+ properties.put(REFRESH_STATUS, resolvedMaterializedTable.getRefreshStatus().name());
+
+ resolvedMaterializedTable
+ .getRefreshHandlerDescription()
+ .ifPresent(
+ refreshHandlerDesc ->
+ properties.put(REFRESH_HANDLER_DESC, refreshHandlerDesc));
+ if (resolvedMaterializedTable.getSerializedRefreshHandler() != null) {
+ properties.put(
+ REFRESH_HANDLER_BYTES,
+ new String(
+ resolvedMaterializedTable.getSerializedRefreshHandler(),
+ StandardCharsets.UTF_8));
+ }
+
+ return properties;
+ } catch (Exception e) {
+ throw new CatalogException("Error in serializing catalog materialized table.", e);
+ }
+ }
+
/** Deserializes the given map of string properties into an unresolved {@link CatalogTable}. */
public static CatalogTable deserializeCatalogTable(Map<String, String> properties) {
return deserializeCatalogTable(properties, null);
@@ -128,7 +181,7 @@ public final class CatalogPropertiesUtil {
* @param properties The properties to deserialize from
* @param fallbackKey The fallback key to get the schema properties. This is meant to support
* the old table (1.10) deserialization
- * @return
+ * @return a catalog table instance.
*/
public static CatalogTable deserializeCatalogTable(
Map<String, String> properties, @Nullable String fallbackKey) {
@@ -158,6 +211,64 @@ public final class CatalogPropertiesUtil {
}
}
+ /**
+ * Deserializes the given map of string properties into an unresolved {@link
+ * CatalogMaterializedTable}.
+ */
+ public static CatalogMaterializedTable deserializeCatalogMaterializedTable(
+ Map<String, String> properties) {
+ try {
+ final Schema schema = deserializeSchema(properties, SCHEMA);
+
+ final @Nullable String comment = properties.get(COMMENT);
+
+ final @Nullable Long snapshot =
+ properties.containsKey(SNAPSHOT)
+ ? getValue(properties, SNAPSHOT, Long::parseLong)
+ : null;
+
+ final List<String> partitionKeys = deserializePartitionKeys(properties);
+
+ final Map<String, String> options = deserializeOptions(properties, SCHEMA);
+
+ final String definitionQuery = properties.get(DEFINITION_QUERY);
+ final Duration freshness = Duration.parse(properties.get(FRESHNESS));
+
+ final CatalogMaterializedTable.LogicalRefreshMode logicalRefreshMode =
+ CatalogMaterializedTable.LogicalRefreshMode.valueOf(
+ properties.get(LOGICAL_REFRESH_MODE));
+ final CatalogMaterializedTable.RefreshMode refreshMode =
+ CatalogMaterializedTable.RefreshMode.valueOf(properties.get(REFRESH_MODE));
+ final CatalogMaterializedTable.RefreshStatus refreshStatus =
+ CatalogMaterializedTable.RefreshStatus.valueOf(properties.get(REFRESH_STATUS));
+
+ final @Nullable String refreshHandlerDesc = properties.get(REFRESH_HANDLER_DESC);
+ final @Nullable String refreshHandlerStringBytes =
+ properties.get(REFRESH_HANDLER_BYTES);
+ final @Nullable byte[] refreshHandlerBytes =
+ StringUtils.isNullOrWhitespaceOnly(refreshHandlerStringBytes)
+ ? null
+ : refreshHandlerStringBytes.getBytes(StandardCharsets.UTF_8);
+
+ CatalogMaterializedTable.Builder builder = CatalogMaterializedTable.newBuilder();
+ builder.schema(schema)
+ .comment(comment)
+ .partitionKeys(partitionKeys)
+ .options(options)
+ .snapshot(snapshot)
+ .definitionQuery(definitionQuery)
+ .freshness(freshness)
+ .logicalRefreshMode(logicalRefreshMode)
+ .refreshMode(refreshMode)
+ .refreshStatus(refreshStatus)
+ .refreshHandlerDescription(refreshHandlerDesc)
+ .serializedRefreshHandler(refreshHandlerBytes);
+ return builder.build();
+ } catch (Exception e) {
+ throw new CatalogException("Error in deserializing catalog materialized table.", e);
+ }
+ }
+
// --------------------------------------------------------------------------------------------
// Helper methods and constants
// --------------------------------------------------------------------------------------------
@@ -205,6 +316,20 @@ public final class CatalogPropertiesUtil {
private static final String SNAPSHOT = "snapshot";
+ private static final String DEFINITION_QUERY = "definition-query";
+
+ private static final String FRESHNESS = "freshness";
+
+ private static final String LOGICAL_REFRESH_MODE = "logical-refresh-mode";
+
+ private static final String REFRESH_MODE = "refresh-mode";
+
+ private static final String REFRESH_STATUS = "refresh-status";
+
+ private static final String REFRESH_HANDLER_DESC = "refresh-handler-desc";
+
+ private static final String REFRESH_HANDLER_BYTES = "refresh-handler-bytes";
+
private static Map<String, String> deserializeOptions(
Map<String, String> map, String schemaKey) {
return map.entrySet().stream()
@@ -214,11 +339,22 @@ public final class CatalogPropertiesUtil {
return !key.startsWith(schemaKey + SEPARATOR)
&& !key.startsWith(PARTITION_KEYS + SEPARATOR)
&& !key.equals(COMMENT)
- && !key.equals(SNAPSHOT);
+ && !key.equals(SNAPSHOT)
+ && !isMaterializedTableAttribute(key);
})
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
}
+ private static boolean isMaterializedTableAttribute(String key) {
+ return key.equals(DEFINITION_QUERY)
+ || key.equals(FRESHNESS)
+ || key.equals(LOGICAL_REFRESH_MODE)
+ || key.equals(REFRESH_MODE)
+ || key.equals(REFRESH_STATUS)
+ || key.equals(REFRESH_HANDLER_DESC)
+ || key.equals(REFRESH_HANDLER_BYTES);
+ }
+
private static List<String> deserializePartitionKeys(Map<String, String> map) {
final int partitionCount = getCount(map, PARTITION_KEYS, NAME);
final List<String> partitionKeys = new ArrayList<>();