You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by ab...@apache.org on 2026/08/26 07:07:26 UTC
(fluss) branch main updated (92f9d48bd -> bc04412ea)
This is an automated email from the ASF dual-hosted git repository.
fresh-borzoni pushed a change to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
from 92f9d48bd [client] Reduce connection timeout to 15s for faster metadata recovery (#4095)
new 01988d8c2 [flink] add support for flink 2.3
new bc04412ea [flink] Fix stale delta join tests and add missing 2.3 coverage
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.github/workflows/stage.sh | 3 +-
.../adapter/CatalogMaterializedTableAdapter.java | 132 +++++++++++++
.../fluss/flink/catalog/Flink22CatalogTest.java | 13 ++
.../{fluss-flink-2.2 => fluss-flink-2.3}/pom.xml | 8 +-
.../adapter/CatalogMaterializedTableAdapter.java | 135 +++++++++++++
.../flink/adapter/IntervalFreshnessAdapter.java} | 41 ++--
.../adapter/MultipleParameterToolAdapter.java | 0
.../apache/fluss/flink/adapter/SchemaAdapter.java | 0
.../apache/fluss/flink/adapter/SinkAdapter.java | 0
.../flink/adapter/TypeInformationAdapter.java | 6 +-
.../src/main/resources/META-INF/NOTICE | 0
.../org.apache.flink.table.factories.Factory | 0
.../org.apache.fluss.flink.action.ActionFactory | 0
.../orphan/Flink23OrphanFilesCleanITCase.java} | 4 +-
.../adapter/Flink23MultipleParameterToolTest.java} | 4 +-
.../fluss/flink/catalog/Flink23CatalogITCase.java} | 4 +-
.../fluss/flink/catalog/Flink23CatalogTest.java | 212 +++++++++++++++++++++
.../catalog/Flink23MaterializedTableITCase.java} | 4 +-
.../flink/catalog/Flink23TableFactoryTest.java} | 4 +-
.../fluss/flink/metrics/Flink23MetricsITCase.java} | 4 +-
.../flink/procedure/Flink23ProcedureITCase.java} | 4 +-
.../security/acl/Flink23AuthorizationITCase.java} | 4 +-
.../flink/sink/Flink23ComplexTypeITCase.java} | 4 +-
.../fluss/flink/sink/Flink23SinkAdapterTest.java | 50 +++++
.../fluss/flink/sink/Flink23TableSinkITCase.java | 26 +--
.../flink/sink/Flink23UndoRecoveryITCase.java} | 4 +-
.../source/Flink23BinlogVirtualTableITCase.java} | 4 +-
.../Flink23ChangelogVirtualTableITCase.java} | 4 +-
.../flink/source/Flink23DeltaJoinITCase.java} | 182 ++++++++++++------
.../source/Flink23TableSourceBatchITCase.java} | 4 +-
.../source/Flink23TableSourceFailOverITCase.java} | 4 +-
.../flink/source/Flink23TableSourceITCase.java} | 4 +-
.../fluss/flink/tiering/Flink23TieringITCase.java} | 4 +-
.../Flink23TieringCommitOperatorTest.java} | 4 +-
.../org.junit.jupiter.api.extension.Extension | 0
.../src/test/resources/junit-platform.properties | 0
.../src/test/resources/log4j2-test.properties | 0
fluss-flink/fluss-flink-common/README.md | 9 +-
.../apache/fluss/flink/FlinkConnectorOptions.java | 24 ++-
.../adapter/CatalogMaterializedTableAdapter.java | 131 +++++++++++++
.../flink/adapter/IntervalFreshnessAdapter.java} | 40 ++--
.../apache/fluss/flink/utils/FlinkConversions.java | 39 +++-
.../fluss/flink/catalog/FlinkCatalogTest.java | 25 ++-
.../fluss/flink/sink/FlinkTableSinkITCase.java | 15 ++
.../fluss/flink/utils/FlinkConversionsTest.java | 139 ++++++++++++++
fluss-flink/pom.xml | 1 +
fluss-test-coverage/pom.xml | 7 +
website/docs/engine-flink/delta-joins.mdx | 27 ++-
website/docs/engine-flink/getting-started.md | 3 +-
49 files changed, 1161 insertions(+), 175 deletions(-)
create mode 100644 fluss-flink/fluss-flink-2.2/src/main/java/org/apache/fluss/flink/adapter/CatalogMaterializedTableAdapter.java
copy fluss-flink/{fluss-flink-2.2 => fluss-flink-2.3}/pom.xml (98%)
create mode 100644 fluss-flink/fluss-flink-2.3/src/main/java/org/apache/fluss/flink/adapter/CatalogMaterializedTableAdapter.java
rename fluss-flink/{fluss-flink-common/src/test/java/org/apache/fluss/flink/adapter/ResolvedCatalogMaterializedTableAdapter.java => fluss-flink-2.3/src/main/java/org/apache/fluss/flink/adapter/IntervalFreshnessAdapter.java} (51%)
copy fluss-flink/{fluss-flink-2.2 => fluss-flink-2.3}/src/main/java/org/apache/fluss/flink/adapter/MultipleParameterToolAdapter.java (100%)
copy fluss-flink/{fluss-flink-2.2 => fluss-flink-2.3}/src/main/java/org/apache/fluss/flink/adapter/SchemaAdapter.java (100%)
copy fluss-flink/{fluss-flink-2.2 => fluss-flink-2.3}/src/main/java/org/apache/fluss/flink/adapter/SinkAdapter.java (100%)
copy fluss-flink/{fluss-flink-2.2 => fluss-flink-2.3}/src/main/java/org/apache/fluss/flink/adapter/TypeInformationAdapter.java (93%)
copy fluss-flink/{fluss-flink-1.18 => fluss-flink-2.3}/src/main/resources/META-INF/NOTICE (100%)
copy fluss-flink/{fluss-flink-2.2 => fluss-flink-2.3}/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory (100%)
copy fluss-flink/{fluss-flink-1.19 => fluss-flink-2.3}/src/main/resources/META-INF/services/org.apache.fluss.flink.action.ActionFactory (100%)
copy fluss-flink/{fluss-flink-2.2/src/test/java/org/apache/fluss/flink/action/orphan/Flink22OrphanFilesCleanITCase.java => fluss-flink-2.3/src/test/java/org/apache/fluss/flink/action/orphan/Flink23OrphanFilesCleanITCase.java} (87%)
copy fluss-flink/{fluss-flink-2.2/src/test/java/org/apache/fluss/flink/adapter/Flink22MultipleParameterToolTest.java => fluss-flink-2.3/src/test/java/org/apache/fluss/flink/adapter/Flink23MultipleParameterToolTest.java} (87%)
copy fluss-flink/{fluss-flink-2.2/src/test/java/org/apache/fluss/flink/catalog/Flink22CatalogITCase.java => fluss-flink-2.3/src/test/java/org/apache/fluss/flink/catalog/Flink23CatalogITCase.java} (98%)
create mode 100644 fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/catalog/Flink23CatalogTest.java
copy fluss-flink/{fluss-flink-2.2/src/test/java/org/apache/fluss/flink/catalog/Flink22MaterializedTableITCase.java => fluss-flink-2.3/src/test/java/org/apache/fluss/flink/catalog/Flink23MaterializedTableITCase.java} (88%)
copy fluss-flink/{fluss-flink-2.2/src/test/java/org/apache/fluss/flink/catalog/Flink22TableFactoryTest.java => fluss-flink-2.3/src/test/java/org/apache/fluss/flink/catalog/Flink23TableFactoryTest.java} (90%)
copy fluss-flink/{fluss-flink-2.2/src/test/java/org/apache/fluss/flink/metrics/Flink22MetricsITCase.java => fluss-flink-2.3/src/test/java/org/apache/fluss/flink/metrics/Flink23MetricsITCase.java} (88%)
copy fluss-flink/{fluss-flink-2.2/src/test/java/org/apache/fluss/flink/procedure/Flink22ProcedureITCase.java => fluss-flink-2.3/src/test/java/org/apache/fluss/flink/procedure/Flink23ProcedureITCase.java} (88%)
copy fluss-flink/{fluss-flink-2.2/src/test/java/org/apache/fluss/flink/security/acl/Flink22AuthorizationITCase.java => fluss-flink-2.3/src/test/java/org/apache/fluss/flink/security/acl/Flink23AuthorizationITCase.java} (88%)
copy fluss-flink/{fluss-flink-2.2/src/test/java/org/apache/fluss/flink/sink/Flink22ComplexTypeITCase.java => fluss-flink-2.3/src/test/java/org/apache/fluss/flink/sink/Flink23ComplexTypeITCase.java} (87%)
create mode 100644 fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/sink/Flink23SinkAdapterTest.java
copy fluss-common/src/main/java/org/apache/fluss/metrics/Gauge.java => fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/sink/Flink23TableSinkITCase.java (55%)
copy fluss-flink/{fluss-flink-1.19/src/test/java/org/apache/fluss/flink/sink/Flink119TableSinkITCase.java => fluss-flink-2.3/src/test/java/org/apache/fluss/flink/sink/Flink23UndoRecoveryITCase.java} (87%)
copy fluss-flink/{fluss-flink-2.2/src/test/java/org/apache/fluss/flink/source/Flink22BinlogVirtualTableITCase.java => fluss-flink-2.3/src/test/java/org/apache/fluss/flink/source/Flink23BinlogVirtualTableITCase.java} (87%)
copy fluss-flink/{fluss-flink-2.2/src/test/java/org/apache/fluss/flink/source/Flink22ChangelogVirtualTableITCase.java => fluss-flink-2.3/src/test/java/org/apache/fluss/flink/source/Flink23ChangelogVirtualTableITCase.java} (87%)
copy fluss-flink/{fluss-flink-2.2/src/test/java/org/apache/fluss/flink/source/Flink22DeltaJoinITCase.java => fluss-flink-2.3/src/test/java/org/apache/fluss/flink/source/Flink23DeltaJoinITCase.java} (90%)
copy fluss-flink/{fluss-flink-2.2/src/test/java/org/apache/fluss/flink/source/Flink22TableSourceBatchITCase.java => fluss-flink-2.3/src/test/java/org/apache/fluss/flink/source/Flink23TableSourceBatchITCase.java} (88%)
copy fluss-flink/{fluss-flink-2.2/src/test/java/org/apache/fluss/flink/source/Flink22TableSourceFailOverITCase.java => fluss-flink-2.3/src/test/java/org/apache/fluss/flink/source/Flink23TableSourceFailOverITCase.java} (87%)
copy fluss-flink/{fluss-flink-2.2/src/test/java/org/apache/fluss/flink/source/Flink22TableSourceITCase.java => fluss-flink-2.3/src/test/java/org/apache/fluss/flink/source/Flink23TableSourceITCase.java} (87%)
copy fluss-flink/{fluss-flink-2.2/src/test/java/org/apache/fluss/flink/tiering/Flink22TieringITCase.java => fluss-flink-2.3/src/test/java/org/apache/fluss/flink/tiering/Flink23TieringITCase.java} (89%)
copy fluss-flink/{fluss-flink-2.2/src/test/java/org/apache/fluss/flink/tiering/committer/Flink22TieringCommitOperatorTest.java => fluss-flink-2.3/src/test/java/org/apache/fluss/flink/tiering/committer/Flink23TieringCommitOperatorTest.java} (92%)
copy {fluss-client => fluss-flink/fluss-flink-2.3}/src/test/resources/META-INF/services/org.junit.jupiter.api.extension.Extension (100%)
copy fluss-flink/{fluss-flink-1.18 => fluss-flink-2.3}/src/test/resources/junit-platform.properties (100%)
copy fluss-flink/{fluss-flink-1.19 => fluss-flink-2.3}/src/test/resources/log4j2-test.properties (100%)
create mode 100644 fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/adapter/CatalogMaterializedTableAdapter.java
rename fluss-flink/{fluss-flink-2.2/src/test/java/org/apache/fluss/flink/adapter/ResolvedCatalogMaterializedTableAdapter.java => fluss-flink-common/src/main/java/org/apache/fluss/flink/adapter/IntervalFreshnessAdapter.java} (50%)
(fluss) 01/02: [flink] add support for flink 2.3
Posted by ab...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
fresh-borzoni pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
commit 01988d8c2b67fddd8dd2966d7ca5e4e8174101b8
Author: Pei Yu <12...@qq.com>
AuthorDate: Tue Aug 25 12:00:17 2026 +0200
[flink] add support for flink 2.3
---
.github/workflows/stage.sh | 3 +-
.../adapter/CatalogMaterializedTableAdapter.java | 132 +++
.../fluss/flink/catalog/Flink22CatalogTest.java | 13 +
fluss-flink/fluss-flink-2.3/pom.xml | 294 ++++++
.../adapter/CatalogMaterializedTableAdapter.java | 135 +++
.../flink/adapter/IntervalFreshnessAdapter.java} | 41 +-
.../adapter/MultipleParameterToolAdapter.java | 68 ++
.../apache/fluss/flink/adapter/SchemaAdapter.java | 43 +
.../apache/fluss/flink/adapter/SinkAdapter.java | 45 +
.../flink/adapter/TypeInformationAdapter.java | 64 ++
.../src/main/resources/META-INF/NOTICE | 9 +
.../org.apache.flink.table.factories.Factory | 19 +
.../org.apache.fluss.flink.action.ActionFactory | 19 +
.../orphan/Flink23OrphanFilesCleanITCase.java | 22 +
.../adapter/Flink23MultipleParameterToolTest.java | 21 +
.../fluss/flink/catalog/Flink23CatalogITCase.java | 178 ++++
.../fluss/flink/catalog/Flink23CatalogTest.java | 196 ++++
.../catalog/Flink23MaterializedTableITCase.java | 21 +
.../flink/catalog/Flink23TableFactoryTest.java | 29 +
.../fluss/flink/metrics/Flink23MetricsITCase.java | 21 +
.../flink/procedure/Flink23ProcedureITCase.java | 21 +
.../security/acl/Flink23AuthorizationITCase.java | 21 +
.../fluss/flink/sink/Flink23ComplexTypeITCase.java | 21 +
.../fluss/flink/sink/Flink23TableSinkITCase.java | 34 +
.../flink/sink/Flink23UndoRecoveryITCase.java | 21 +
.../source/Flink23BinlogVirtualTableITCase.java | 21 +
.../source/Flink23ChangelogVirtualTableITCase.java | 21 +
.../fluss/flink/source/Flink23DeltaJoinITCase.java | 1101 ++++++++++++++++++++
.../source/Flink23TableSourceBatchITCase.java | 21 +
.../source/Flink23TableSourceFailOverITCase.java | 21 +
.../flink/source/Flink23TableSourceITCase.java | 21 +
.../fluss/flink/tiering/Flink23TieringITCase.java | 22 +
.../Flink23TieringCommitOperatorTest.java | 24 +
.../org.junit.jupiter.api.extension.Extension | 19 +
.../src/test/resources/junit-platform.properties | 20 +
.../src/test/resources/log4j2-test.properties | 32 +
fluss-flink/fluss-flink-common/README.md | 9 +-
.../apache/fluss/flink/FlinkConnectorOptions.java | 24 +-
.../adapter/CatalogMaterializedTableAdapter.java | 131 +++
.../flink/adapter/IntervalFreshnessAdapter.java} | 40 +-
.../apache/fluss/flink/utils/FlinkConversions.java | 39 +-
.../fluss/flink/catalog/FlinkCatalogTest.java | 25 +-
.../fluss/flink/sink/FlinkTableSinkITCase.java | 15 +
.../fluss/flink/utils/FlinkConversionsTest.java | 139 +++
fluss-flink/pom.xml | 1 +
fluss-test-coverage/pom.xml | 7 +
website/docs/engine-flink/delta-joins.mdx | 27 +-
website/docs/engine-flink/getting-started.md | 3 +-
48 files changed, 3212 insertions(+), 62 deletions(-)
diff --git a/.github/workflows/stage.sh b/.github/workflows/stage.sh
index de2c7768b..c302f0f4d 100755
--- a/.github/workflows/stage.sh
+++ b/.github/workflows/stage.sh
@@ -36,7 +36,8 @@ fluss-flink/fluss-flink-1.18\
MODULES_FLINK2="\
fluss-flink,\
fluss-flink/fluss-flink-common,\
-fluss-flink/fluss-flink-2.2\
+fluss-flink/fluss-flink-2.2,\
+fluss-flink/fluss-flink-2.3\
"
MODULES_COMMON_SPARK="\
diff --git a/fluss-flink/fluss-flink-2.2/src/main/java/org/apache/fluss/flink/adapter/CatalogMaterializedTableAdapter.java b/fluss-flink/fluss-flink-2.2/src/main/java/org/apache/fluss/flink/adapter/CatalogMaterializedTableAdapter.java
new file mode 100644
index 000000000..a64581a72
--- /dev/null
+++ b/fluss-flink/fluss-flink-2.2/src/main/java/org/apache/fluss/flink/adapter/CatalogMaterializedTableAdapter.java
@@ -0,0 +1,132 @@
+/*
+ * 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.fluss.flink.adapter;
+
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.catalog.CatalogMaterializedTable;
+import org.apache.flink.table.catalog.IntervalFreshness;
+import org.apache.flink.table.catalog.TableDistribution;
+
+import javax.annotation.Nullable;
+
+import java.util.List;
+import java.util.Map;
+
+/** An adapter for {@link CatalogMaterializedTable#newBuilder()} constructor for flink2.2. */
+public class CatalogMaterializedTableAdapter {
+
+ private final CatalogMaterializedTable.Builder builder;
+
+ private CatalogMaterializedTableAdapter() {
+ this.builder = CatalogMaterializedTable.newBuilder();
+ }
+
+ public static CatalogMaterializedTableAdapter newAdapter() {
+ return new CatalogMaterializedTableAdapter();
+ }
+
+ public CatalogMaterializedTableAdapter schema(Schema schema) {
+ this.builder.schema(schema);
+ return this;
+ }
+
+ public CatalogMaterializedTableAdapter comment(@Nullable String comment) {
+ this.builder.comment(comment);
+ return this;
+ }
+
+ public CatalogMaterializedTableAdapter partitionKeys(List<String> partitionKeys) {
+ this.builder.partitionKeys(partitionKeys);
+ return this;
+ }
+
+ public CatalogMaterializedTableAdapter options(Map<String, String> options) {
+ this.builder.options(options);
+ return this;
+ }
+
+ public CatalogMaterializedTableAdapter snapshot(@Nullable Long snapshot) {
+ this.builder.snapshot(snapshot);
+ return this;
+ }
+
+ public CatalogMaterializedTableAdapter originalQuery(String originalQuery) {
+ return this;
+ }
+
+ public CatalogMaterializedTableAdapter expandedQuery(String expandedQuery) {
+ return this;
+ }
+
+ public CatalogMaterializedTableAdapter definitionQuery(String definitionQuery) {
+ this.builder.definitionQuery(definitionQuery);
+ return this;
+ }
+
+ public CatalogMaterializedTableAdapter freshness(@Nullable IntervalFreshness freshness) {
+ this.builder.freshness(freshness);
+ return this;
+ }
+
+ public CatalogMaterializedTableAdapter logicalRefreshMode(
+ CatalogMaterializedTable.LogicalRefreshMode logicalRefreshMode) {
+ this.builder.logicalRefreshMode(logicalRefreshMode);
+ return this;
+ }
+
+ public CatalogMaterializedTableAdapter refreshMode(
+ @Nullable CatalogMaterializedTable.RefreshMode refreshMode) {
+ this.builder.refreshMode(refreshMode);
+ return this;
+ }
+
+ public CatalogMaterializedTableAdapter refreshStatus(
+ CatalogMaterializedTable.RefreshStatus refreshStatus) {
+ this.builder.refreshStatus(refreshStatus);
+ return this;
+ }
+
+ public CatalogMaterializedTableAdapter refreshHandlerDescription(
+ @Nullable String refreshHandlerDescription) {
+ this.builder.refreshHandlerDescription(refreshHandlerDescription);
+ return this;
+ }
+
+ public CatalogMaterializedTableAdapter serializedRefreshHandler(
+ @Nullable byte[] serializedRefreshHandler) {
+ this.builder.serializedRefreshHandler(serializedRefreshHandler);
+ return this;
+ }
+
+ public CatalogMaterializedTableAdapter distribution(@Nullable TableDistribution distribution) {
+ this.builder.distribution(distribution);
+ return this;
+ }
+
+ public CatalogMaterializedTable build() {
+ return this.builder.build();
+ }
+
+ public static String getExpandedQuery(CatalogMaterializedTable mt) {
+ return null;
+ }
+
+ public static String getOriginalQuery(CatalogMaterializedTable mt) {
+ return null;
+ }
+}
diff --git a/fluss-flink/fluss-flink-2.2/src/test/java/org/apache/fluss/flink/catalog/Flink22CatalogTest.java b/fluss-flink/fluss-flink-2.2/src/test/java/org/apache/fluss/flink/catalog/Flink22CatalogTest.java
index bff2b77d6..0c49cc713 100644
--- a/fluss-flink/fluss-flink-2.2/src/test/java/org/apache/fluss/flink/catalog/Flink22CatalogTest.java
+++ b/fluss-flink/fluss-flink-2.2/src/test/java/org/apache/fluss/flink/catalog/Flink22CatalogTest.java
@@ -18,8 +18,11 @@
package org.apache.fluss.flink.catalog;
import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.catalog.CatalogMaterializedTable;
import org.apache.flink.table.catalog.Column;
import org.apache.flink.table.catalog.DefaultIndex;
+import org.apache.flink.table.catalog.IntervalFreshness;
+import org.apache.flink.table.catalog.ResolvedCatalogMaterializedTable;
import org.apache.flink.table.catalog.ResolvedSchema;
import org.apache.flink.table.catalog.UniqueConstraint;
@@ -41,4 +44,14 @@ public class Flink22CatalogTest extends FlinkCatalogTest {
DefaultIndex.newIndex(
"INDEX_first_third", Arrays.asList("first", "third"))));
}
+
+ @Override
+ protected ResolvedCatalogMaterializedTable createResolvedCatalogMaterializedTable(
+ CatalogMaterializedTable origin,
+ ResolvedSchema resolvedSchema,
+ CatalogMaterializedTable.RefreshMode refreshMode,
+ IntervalFreshness intervalFreshness) {
+ return new ResolvedCatalogMaterializedTable(
+ origin, resolvedSchema, refreshMode, intervalFreshness);
+ }
}
diff --git a/fluss-flink/fluss-flink-2.3/pom.xml b/fluss-flink/fluss-flink-2.3/pom.xml
new file mode 100644
index 000000000..1475fb3e0
--- /dev/null
+++ b/fluss-flink/fluss-flink-2.3/pom.xml
@@ -0,0 +1,294 @@
+<?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.fluss</groupId>
+ <artifactId>fluss-flink</artifactId>
+ <version>1.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>fluss-flink-2.3</artifactId>
+ <name>Apache Fluss : Engine Flink : 2.3 </name>
+ <properties>
+ <flink.major.version>2.3</flink.major.version>
+ <flink.minor.version>2.3.0</flink.minor.version>
+ </properties>
+
+ <dependencies>
+ <!-- Fluss dependency -->
+ <dependency>
+ <groupId>org.apache.fluss</groupId>
+ <artifactId>fluss-client</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.fluss</groupId>
+ <artifactId>fluss-flink-common</artifactId>
+ <version>${project.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>*</groupId>
+ <artifactId>*</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+
+ <!-- Flink dependency -->
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-core</artifactId>
+ <version>${flink.minor.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-common</artifactId>
+ <version>${flink.minor.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-streaming-java</artifactId>
+ <version>${flink.minor.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-runtime</artifactId>
+ <version>${flink.minor.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <!-- test dependency -->
+ <dependency>
+ <groupId>org.apache.fluss</groupId>
+ <artifactId>fluss-flink-common</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ <type>test-jar</type>
+ </dependency>
+
+ <dependency>
+ <groupId>org.roaringbitmap</groupId>
+ <artifactId>RoaringBitmap</artifactId>
+ <version>${roaringbitmap.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.fluss</groupId>
+ <artifactId>fluss-server</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-test-utils</artifactId>
+ <version>${flink.minor.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-base</artifactId>
+ <version>${flink.minor.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.fluss</groupId>
+ <artifactId>fluss-server</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ <type>test-jar</type>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.fluss</groupId>
+ <artifactId>fluss-rpc</artifactId>
+ <scope>test</scope>
+ <type>test-jar</type>
+ <version>${project.version}</version>
+ </dependency>
+
+ <!-- for curator TestingServer -->
+ <dependency>
+ <groupId>org.apache.curator</groupId>
+ <artifactId>curator-test</artifactId>
+ <version>${curator.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-common</artifactId>
+ <version>${flink.minor.version}</version>
+ <scope>test</scope>
+ <type>test-jar</type>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-test-utils</artifactId>
+ <version>${flink.minor.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+
+ <dependency>
+ <groupId>org.apache.fluss</groupId>
+ <artifactId>fluss-test-utils</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.fluss</groupId>
+ <artifactId>fluss-common</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ <type>test-jar</type>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-clients</artifactId>
+ <version>${flink.minor.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-sql-gateway</artifactId>
+ <version>${flink.minor.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-sql-gateway</artifactId>
+ <version>${flink.minor.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-compiler-plugin</artifactId>
+ <configuration>
+ <!-- compilation of main sources -->
+ <skipMain>${skip.on.java8}</skipMain>
+ <!-- compilation of test sources -->
+ <skip>${skip.on.java8}</skip>
+ </configuration>
+ </plugin>
+
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <version>3.0.0-M5</version>
+ <executions>
+ <!-- Test end with ITCase is e2e test in this module -->
+ <execution>
+ <id>integration-tests</id>
+ <phase>integration-test</phase>
+ <inherited>false</inherited>
+ <goals>
+ <goal>test</goal>
+ </goals>
+ <configuration>
+ <skip>${skip.on.java8}</skip>
+ <includes>
+ <include>**/*ITCase.*</include>
+ </includes>
+ <!-- e2e test with flink/zookeeper cluster, we set forkCount=1 -->
+ <forkCount>1</forkCount>
+ </configuration>
+ </execution>
+ <!-- others unit tests -->
+ <execution>
+ <id>default-test</id>
+ <phase>test</phase>
+ <inherited>false</inherited>
+ <goals>
+ <goal>test</goal>
+ </goals>
+ <configuration>
+ <skip>${skip.on.java8}</skip>
+ <excludes>
+ <exclude>**/*ITCase.*</exclude>
+ </excludes>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>shade-fluss</id>
+ <phase>package</phase>
+ <goals>
+ <goal>shade</goal>
+ </goals>
+ <configuration>
+ <artifactSet>
+ <includes combine.children="append">
+ <include>org.apache.fluss:fluss-flink-common</include>
+ <include>org.apache.fluss:fluss-client</include>
+ <include>org.roaringbitmap:RoaringBitmap</include>
+ </includes>
+ </artifactSet>
+ <relocations combine.children="append">
+ <relocation>
+ <pattern>org.roaringbitmap</pattern>
+ <shadedPattern>org.apache.fluss.shaded.org.roaringbitmap</shadedPattern>
+ </relocation>
+ </relocations>
+ <filters combine.children="append">
+ <filter>
+ <artifact>org.roaringbitmap:RoaringBitmap</artifact>
+ <excludes>
+ <exclude>META-INF/versions/**</exclude>
+ </excludes>
+ </filter>
+ </filters>
+ <transformers combine.children="append">
+ <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
+ <mainClass>org.apache.fluss.flink.action.FlussActionEntrypoint</mainClass>
+ </transformer>
+ </transformers>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+
+</project>
diff --git a/fluss-flink/fluss-flink-2.3/src/main/java/org/apache/fluss/flink/adapter/CatalogMaterializedTableAdapter.java b/fluss-flink/fluss-flink-2.3/src/main/java/org/apache/fluss/flink/adapter/CatalogMaterializedTableAdapter.java
new file mode 100644
index 000000000..6aa91eeda
--- /dev/null
+++ b/fluss-flink/fluss-flink-2.3/src/main/java/org/apache/fluss/flink/adapter/CatalogMaterializedTableAdapter.java
@@ -0,0 +1,135 @@
+/*
+ * 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.fluss.flink.adapter;
+
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.catalog.CatalogMaterializedTable;
+import org.apache.flink.table.catalog.IntervalFreshness;
+import org.apache.flink.table.catalog.TableDistribution;
+
+import javax.annotation.Nullable;
+
+import java.util.List;
+import java.util.Map;
+
+/** An adapter for {@link CatalogMaterializedTable#newBuilder()} constructor for flink2.3. */
+public class CatalogMaterializedTableAdapter {
+
+ private final CatalogMaterializedTable.Builder builder;
+
+ private CatalogMaterializedTableAdapter() {
+ this.builder = CatalogMaterializedTable.newBuilder();
+ }
+
+ public static CatalogMaterializedTableAdapter newAdapter() {
+ return new CatalogMaterializedTableAdapter();
+ }
+
+ public CatalogMaterializedTableAdapter schema(Schema schema) {
+ this.builder.schema(schema);
+ return this;
+ }
+
+ public CatalogMaterializedTableAdapter comment(@Nullable String comment) {
+ this.builder.comment(comment);
+ return this;
+ }
+
+ public CatalogMaterializedTableAdapter partitionKeys(List<String> partitionKeys) {
+ this.builder.partitionKeys(partitionKeys);
+ return this;
+ }
+
+ public CatalogMaterializedTableAdapter options(Map<String, String> options) {
+ this.builder.options(options);
+ return this;
+ }
+
+ public CatalogMaterializedTableAdapter snapshot(@Nullable Long snapshot) {
+ this.builder.snapshot(snapshot);
+ return this;
+ }
+
+ public CatalogMaterializedTableAdapter originalQuery(String originalQuery) {
+ this.builder.originalQuery(originalQuery);
+ return this;
+ }
+
+ public CatalogMaterializedTableAdapter expandedQuery(String expandedQuery) {
+ this.builder.expandedQuery(expandedQuery);
+ return this;
+ }
+
+ @Deprecated
+ public CatalogMaterializedTableAdapter definitionQuery(String definitionQuery) {
+ this.builder.definitionQuery(definitionQuery);
+ return this;
+ }
+
+ public CatalogMaterializedTableAdapter freshness(@Nullable IntervalFreshness freshness) {
+ this.builder.freshness(freshness);
+ return this;
+ }
+
+ public CatalogMaterializedTableAdapter logicalRefreshMode(
+ CatalogMaterializedTable.LogicalRefreshMode logicalRefreshMode) {
+ this.builder.logicalRefreshMode(logicalRefreshMode);
+ return this;
+ }
+
+ public CatalogMaterializedTableAdapter refreshMode(
+ @Nullable CatalogMaterializedTable.RefreshMode refreshMode) {
+ this.builder.refreshMode(refreshMode);
+ return this;
+ }
+
+ public CatalogMaterializedTableAdapter refreshStatus(
+ CatalogMaterializedTable.RefreshStatus refreshStatus) {
+ this.builder.refreshStatus(refreshStatus);
+ return this;
+ }
+
+ public CatalogMaterializedTableAdapter refreshHandlerDescription(
+ @Nullable String refreshHandlerDescription) {
+ this.builder.refreshHandlerDescription(refreshHandlerDescription);
+ return this;
+ }
+
+ public CatalogMaterializedTableAdapter serializedRefreshHandler(
+ @Nullable byte[] serializedRefreshHandler) {
+ this.builder.serializedRefreshHandler(serializedRefreshHandler);
+ return this;
+ }
+
+ public CatalogMaterializedTableAdapter distribution(@Nullable TableDistribution distribution) {
+ this.builder.distribution(distribution);
+ return this;
+ }
+
+ public CatalogMaterializedTable build() {
+ return this.builder.build();
+ }
+
+ public static String getExpandedQuery(CatalogMaterializedTable mt) {
+ return mt.getExpandedQuery();
+ }
+
+ public static String getOriginalQuery(CatalogMaterializedTable mt) {
+ return mt.getOriginalQuery();
+ }
+}
diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/adapter/ResolvedCatalogMaterializedTableAdapter.java b/fluss-flink/fluss-flink-2.3/src/main/java/org/apache/fluss/flink/adapter/IntervalFreshnessAdapter.java
similarity index 51%
rename from fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/adapter/ResolvedCatalogMaterializedTableAdapter.java
rename to fluss-flink/fluss-flink-2.3/src/main/java/org/apache/fluss/flink/adapter/IntervalFreshnessAdapter.java
index 7a018bc02..cc747d12e 100644
--- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/adapter/ResolvedCatalogMaterializedTableAdapter.java
+++ b/fluss-flink/fluss-flink-2.3/src/main/java/org/apache/fluss/flink/adapter/IntervalFreshnessAdapter.java
@@ -17,25 +17,30 @@
package org.apache.fluss.flink.adapter;
-import org.apache.flink.table.catalog.CatalogMaterializedTable;
+import org.apache.flink.table.catalog.Interval;
import org.apache.flink.table.catalog.IntervalFreshness;
-import org.apache.flink.table.catalog.ResolvedCatalogMaterializedTable;
-import org.apache.flink.table.catalog.ResolvedSchema;
-/**
- * Adapter for {@link ResolvedCatalogMaterializedTable} because the constructor is compatibility in
- * flink 2.2. However, this constructor only used in test.
- *
- * <p>TODO: remove it until <a href="https://issues.apache.org/jira/browse/FLINK-38532">...</a> is
- * fixed.
- */
-public class ResolvedCatalogMaterializedTableAdapter {
-
- public static ResolvedCatalogMaterializedTable create(
- CatalogMaterializedTable origin,
- ResolvedSchema resolvedSchema,
- CatalogMaterializedTable.RefreshMode refreshMode,
- IntervalFreshness freshness) {
- return new ResolvedCatalogMaterializedTable(origin, resolvedSchema);
+/** An adapter for {@link IntervalFreshness} for Flink 2.3. */
+public class IntervalFreshnessAdapter {
+
+ public static TimeUnitAdapter timeUnit(String name) {
+ return new TimeUnitAdapter(Interval.TimeUnit.valueOf(name));
+ }
+
+ public static IntervalFreshness of(String interval, TimeUnitAdapter timeUnit) {
+ return IntervalFreshness.of(interval, timeUnit.timeUnit);
+ }
+
+ public static String getTimeUnitName(IntervalFreshness intervalFreshness) {
+ return intervalFreshness.getTimeUnit().name();
+ }
+
+ /** An adapter for {@link Interval.TimeUnit} for Flink 2.3. */
+ public static class TimeUnitAdapter {
+ private final Interval.TimeUnit timeUnit;
+
+ private TimeUnitAdapter(Interval.TimeUnit timeUnit) {
+ this.timeUnit = timeUnit;
+ }
}
}
diff --git a/fluss-flink/fluss-flink-2.3/src/main/java/org/apache/fluss/flink/adapter/MultipleParameterToolAdapter.java b/fluss-flink/fluss-flink-2.3/src/main/java/org/apache/fluss/flink/adapter/MultipleParameterToolAdapter.java
new file mode 100644
index 000000000..999cd40a8
--- /dev/null
+++ b/fluss-flink/fluss-flink-2.3/src/main/java/org/apache/fluss/flink/adapter/MultipleParameterToolAdapter.java
@@ -0,0 +1,68 @@
+/*
+ * 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.fluss.flink.adapter;
+
+import org.apache.flink.util.MultipleParameterTool;
+
+import javax.annotation.Nullable;
+
+import java.util.Collection;
+import java.util.Map;
+
+/**
+ * An adapter for Flink {@link MultipleParameterTool} class. The {@link MultipleParameterTool} is
+ * moved to a new package since Flink 2.x, so this adapter helps to bridge compatibility for
+ * different Flink versions.
+ *
+ * <p>TODO: remove this class when no longer support all the Flink 1.x series.
+ */
+public class MultipleParameterToolAdapter {
+
+ private MultipleParameterToolAdapter() {}
+
+ private MultipleParameterTool multipleParameterTool;
+
+ public static MultipleParameterToolAdapter fromArgs(String[] args) {
+ MultipleParameterToolAdapter adapter = new MultipleParameterToolAdapter();
+ adapter.multipleParameterTool = MultipleParameterTool.fromArgs(args);
+ return adapter;
+ }
+
+ public Map<String, String> toMap() {
+ return this.multipleParameterTool.toMap();
+ }
+
+ /** Returns whether the given key is present in the parsed arguments. */
+ public boolean has(String key) {
+ return this.multipleParameterTool.has(key);
+ }
+
+ /** Returns the value for the given key, or {@code null} if the key is not found. */
+ @Nullable
+ public String get(String key) {
+ return this.multipleParameterTool.get(key);
+ }
+
+ /**
+ * Returns all values associated with the given key, or {@code null} if the key is not found.
+ */
+ @Nullable
+ public Collection<String> getMultiParameter(String key) {
+ return this.multipleParameterTool.getMultiParameter(key);
+ }
+}
diff --git a/fluss-flink/fluss-flink-2.3/src/main/java/org/apache/fluss/flink/adapter/SchemaAdapter.java b/fluss-flink/fluss-flink-2.3/src/main/java/org/apache/fluss/flink/adapter/SchemaAdapter.java
new file mode 100644
index 000000000..8f6492bd1
--- /dev/null
+++ b/fluss-flink/fluss-flink-2.3/src/main/java/org/apache/fluss/flink/adapter/SchemaAdapter.java
@@ -0,0 +1,43 @@
+/*
+ * 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.fluss.flink.adapter;
+
+import org.apache.flink.table.api.Schema;
+
+import java.util.List;
+
+/**
+ * An adapter for the schema with Index.
+ *
+ * <p>TODO: remove this class when no longer support all the Flink 1.x series.
+ */
+public class SchemaAdapter {
+ private SchemaAdapter() {}
+
+ public static Schema withIndex(Schema unresolvedSchema, List<List<String>> indexes) {
+ Schema.Builder newSchemaBuilder = Schema.newBuilder().fromSchema(unresolvedSchema);
+ for (List<String> index : indexes) {
+ newSchemaBuilder.index(index);
+ }
+ return newSchemaBuilder.build();
+ }
+
+ public static boolean supportIndex() {
+ return true;
+ }
+}
diff --git a/fluss-flink/fluss-flink-2.3/src/main/java/org/apache/fluss/flink/adapter/SinkAdapter.java b/fluss-flink/fluss-flink-2.3/src/main/java/org/apache/fluss/flink/adapter/SinkAdapter.java
new file mode 100644
index 000000000..76893b443
--- /dev/null
+++ b/fluss-flink/fluss-flink-2.3/src/main/java/org/apache/fluss/flink/adapter/SinkAdapter.java
@@ -0,0 +1,45 @@
+/*
+ * 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.fluss.flink.adapter;
+
+import org.apache.flink.api.common.operators.MailboxExecutor;
+import org.apache.flink.api.connector.sink2.Sink;
+import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.flink.api.connector.sink2.WriterInitContext;
+import org.apache.flink.metrics.groups.SinkWriterMetricGroup;
+
+import java.io.IOException;
+
+/**
+ * Flink sink adapter which hide the different version of createWriter method.
+ *
+ * <p>TODO: remove this class when no longer support all the Flink 1.x series.
+ */
+public abstract class SinkAdapter<InputT> implements Sink<InputT> {
+
+ @Override
+ public SinkWriter<InputT> createWriter(WriterInitContext writerInitContext) throws IOException {
+ return createWriter(
+ writerInitContext.getMailboxExecutor(),
+ writerInitContext.metricGroup(),
+ writerInitContext.getTaskInfo().getIndexOfThisSubtask());
+ }
+
+ protected abstract SinkWriter<InputT> createWriter(
+ MailboxExecutor mailboxExecutor, SinkWriterMetricGroup metricGroup, int subtaskIndex);
+}
diff --git a/fluss-flink/fluss-flink-2.3/src/main/java/org/apache/fluss/flink/adapter/TypeInformationAdapter.java b/fluss-flink/fluss-flink-2.3/src/main/java/org/apache/fluss/flink/adapter/TypeInformationAdapter.java
new file mode 100644
index 000000000..ad4f61145
--- /dev/null
+++ b/fluss-flink/fluss-flink-2.3/src/main/java/org/apache/fluss/flink/adapter/TypeInformationAdapter.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.fluss.flink.adapter;
+
+import org.apache.flink.api.common.serialization.SerializerConfig;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+
+/**
+ * Flink 2.3 variant of the type information adapter.
+ *
+ * <p>In Flink 2.3, {@link TypeInformation} declares {@code createSerializer(SerializerConfig)} as
+ * the abstract method. This adapter implements that method and delegates to {@link
+ * #createSerializer(TypeSerializerCreator)} with a creator that forwards the config, so subclasses
+ * (e.g. in fluss-flink-common) can obtain the inner serializer and wrap it without touching
+ * SerializerConfig here.
+ *
+ * <p>When building for Flink 2.3, this class is used instead of the common adapter so that the
+ * correct API is implemented for this Flink version. See fluss-flink-common's
+ * TypeInformationAdapter for the version that implements both createSerializer(SerializerConfig)
+ * and createSerializer(ExecutionConfig).
+ *
+ * @param <T> the type described by this type information
+ */
+public abstract class TypeInformationAdapter<T> extends TypeInformation<T> {
+
+ @Override
+ public TypeSerializer<T> createSerializer(SerializerConfig config) {
+ return createSerializer(typeInfo -> typeInfo.createSerializer(config));
+ }
+
+ /**
+ * Creates the type serializer using the given creator. The creator captures the config from the
+ * framework; subclasses call {@code creator.createSerializer(innerTypeInfo)} to obtain the
+ * inner serializer and then wrap or return it.
+ */
+ protected abstract TypeSerializer<T> createSerializer(
+ TypeInformationAdapter.TypeSerializerCreator typeSerializerCreator);
+
+ /**
+ * Creator that, given a TypeInformation, returns its serializer for the current config. Passed
+ * by the adapter so that config is forwarded to the underlying TypeInformation.
+ */
+ @FunctionalInterface
+ public interface TypeSerializerCreator {
+ TypeSerializer<?> createSerializer(TypeInformation<?> typeInfo);
+ }
+}
diff --git a/fluss-flink/fluss-flink-2.3/src/main/resources/META-INF/NOTICE b/fluss-flink/fluss-flink-2.3/src/main/resources/META-INF/NOTICE
new file mode 100644
index 000000000..011601daa
--- /dev/null
+++ b/fluss-flink/fluss-flink-2.3/src/main/resources/META-INF/NOTICE
@@ -0,0 +1,9 @@
+fluss-flink-connector
+Copyright 2024-2026 The Apache Software Foundation
+
+This project includes software developed at
+The Apache Software Foundation (http://www.apache.org/).
+
+This project bundles the following dependencies under the Apache Software License 2.0 (http://www.apache.org/licenses/LICENSE-2.0.txt)
+
+- org.roaringbitmap:RoaringBitmap:1.3.0
diff --git a/fluss-flink/fluss-flink-2.3/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory b/fluss-flink/fluss-flink-2.3/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
new file mode 100644
index 000000000..d5aca2d53
--- /dev/null
+++ b/fluss-flink/fluss-flink-2.3/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
@@ -0,0 +1,19 @@
+#
+# 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.fluss.flink.catalog.FlinkCatalogFactory
diff --git a/fluss-flink/fluss-flink-2.3/src/main/resources/META-INF/services/org.apache.fluss.flink.action.ActionFactory b/fluss-flink/fluss-flink-2.3/src/main/resources/META-INF/services/org.apache.fluss.flink.action.ActionFactory
new file mode 100644
index 000000000..c30c9dd5a
--- /dev/null
+++ b/fluss-flink/fluss-flink-2.3/src/main/resources/META-INF/services/org.apache.fluss.flink.action.ActionFactory
@@ -0,0 +1,19 @@
+#
+# 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.fluss.flink.action.orphan.OrphanFilesCleanActionFactory
diff --git a/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/action/orphan/Flink23OrphanFilesCleanITCase.java b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/action/orphan/Flink23OrphanFilesCleanITCase.java
new file mode 100644
index 000000000..1a33148f5
--- /dev/null
+++ b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/action/orphan/Flink23OrphanFilesCleanITCase.java
@@ -0,0 +1,22 @@
+/*
+ * 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.fluss.flink.action.orphan;
+
+/** The IT case for orphan files cleanup in Flink 2.3. */
+class Flink23OrphanFilesCleanITCase extends OrphanFilesCleanITCase {}
diff --git a/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/adapter/Flink23MultipleParameterToolTest.java b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/adapter/Flink23MultipleParameterToolTest.java
new file mode 100644
index 000000000..3cdd2687f
--- /dev/null
+++ b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/adapter/Flink23MultipleParameterToolTest.java
@@ -0,0 +1,21 @@
+/*
+ * 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.fluss.flink.adapter;
+
+/** Test for {@link MultipleParameterToolAdapter} in flink 2.3. */
+public class Flink23MultipleParameterToolTest extends FlinkMultipleParameterToolTest {}
diff --git a/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/catalog/Flink23CatalogITCase.java b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/catalog/Flink23CatalogITCase.java
new file mode 100644
index 000000000..f7e9878d3
--- /dev/null
+++ b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/catalog/Flink23CatalogITCase.java
@@ -0,0 +1,178 @@
+/*
+ * 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.fluss.flink.catalog;
+
+import org.apache.fluss.testutils.common.MultiVersionTest;
+
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** IT case for catalog in Flink 2.3. */
+public class Flink23CatalogITCase extends FlinkCatalogITCase {
+
+ @Test
+ @MultiVersionTest
+ void testGetTableWithIndex() throws Exception {
+ String tableName = "table_with_pk_only";
+ tEnv.executeSql(
+ String.format(
+ "create table %s ( "
+ + " a int, "
+ + " b varchar, "
+ + " c bigint, "
+ + " primary key (a, b) NOT ENFORCED"
+ + ") with ( "
+ + " 'connector' = 'fluss' "
+ + ")",
+ tableName));
+ CatalogTable table = (CatalogTable) catalog.getTable(new ObjectPath(DEFAULT_DB, tableName));
+ Schema expectedSchema =
+ Schema.newBuilder()
+ .column("a", DataTypes.INT().notNull())
+ .column("b", DataTypes.STRING().notNull())
+ .column("c", DataTypes.BIGINT())
+ .primaryKey("a", "b")
+ .index("a", "b")
+ .build();
+ assertThat(table.getUnresolvedSchema()).isEqualTo(expectedSchema);
+
+ tableName = "table_with_prefix_bucket_key";
+ tEnv.executeSql(
+ String.format(
+ "create table %s ( "
+ + " a int, "
+ + " b varchar, "
+ + " c bigint, "
+ + " primary key (a, b) NOT ENFORCED"
+ + ") with ( "
+ + " 'connector' = 'fluss', "
+ + " 'bucket.key' = 'a'"
+ + ")",
+ tableName));
+
+ table = (CatalogTable) catalog.getTable(new ObjectPath(DEFAULT_DB, tableName));
+ expectedSchema =
+ Schema.newBuilder()
+ .column("a", DataTypes.INT().notNull())
+ .column("b", DataTypes.STRING().notNull())
+ .column("c", DataTypes.BIGINT())
+ .primaryKey("a", "b")
+ .index("a", "b")
+ .index("a")
+ .build();
+ assertThat(table.getUnresolvedSchema()).isEqualTo(expectedSchema);
+
+ tableName = "table_with_bucket_key_is_not_prefix_pk";
+ tEnv.executeSql(
+ String.format(
+ "create table %s ( "
+ + " a int, "
+ + " b varchar, "
+ + " c bigint, "
+ + " primary key (a, b) NOT ENFORCED"
+ + ") with ( "
+ + " 'connector' = 'fluss', "
+ + " 'bucket.key' = 'b'"
+ + ")",
+ tableName));
+
+ table = (CatalogTable) catalog.getTable(new ObjectPath(DEFAULT_DB, tableName));
+ expectedSchema =
+ Schema.newBuilder()
+ .column("a", DataTypes.INT().notNull())
+ .column("b", DataTypes.STRING().notNull())
+ .column("c", DataTypes.BIGINT())
+ .primaryKey("a", "b")
+ .index("a", "b")
+ .build();
+ assertThat(table.getUnresolvedSchema()).isEqualTo(expectedSchema);
+
+ tableName = "table_with_partition_1";
+ tEnv.executeSql(
+ String.format(
+ "create table %s ( "
+ + " a int, "
+ + " b varchar, "
+ + " c bigint, "
+ + " dt varchar, "
+ + " primary key (a, b, dt) NOT ENFORCED "
+ + ") "
+ + " partitioned by (dt) "
+ + " with ( "
+ + " 'connector' = 'fluss', "
+ + " 'bucket.key' = 'a'"
+ + ")",
+ tableName));
+
+ table = (CatalogTable) catalog.getTable(new ObjectPath(DEFAULT_DB, tableName));
+ expectedSchema =
+ Schema.newBuilder()
+ .column("a", DataTypes.INT().notNull())
+ .column("b", DataTypes.STRING().notNull())
+ .column("c", DataTypes.BIGINT())
+ .column("dt", DataTypes.STRING().notNull())
+ .primaryKey("a", "b", "dt")
+ .index("a", "b", "dt")
+ .index("a", "dt")
+ .build();
+ assertThat(table.getUnresolvedSchema()).isEqualTo(expectedSchema);
+
+ tableName = "table_with_partition_2";
+ tEnv.executeSql(
+ String.format(
+ "create table %s ( "
+ + " a int, "
+ + " b varchar, "
+ + " c bigint, "
+ + " dt varchar, "
+ + " primary key (dt, a, b) NOT ENFORCED "
+ + ") "
+ + " partitioned by (dt) "
+ + " with ( "
+ + " 'connector' = 'fluss', "
+ + " 'bucket.key' = 'a'"
+ + ")",
+ tableName));
+
+ table = (CatalogTable) catalog.getTable(new ObjectPath(DEFAULT_DB, tableName));
+ expectedSchema =
+ Schema.newBuilder()
+ .column("a", DataTypes.INT().notNull())
+ .column("b", DataTypes.STRING().notNull())
+ .column("c", DataTypes.BIGINT())
+ .column("dt", DataTypes.STRING().notNull())
+ .primaryKey("dt", "a", "b")
+ .index("dt", "a", "b")
+ .index("a", "dt")
+ .build();
+ assertThat(table.getUnresolvedSchema()).isEqualTo(expectedSchema);
+ }
+
+ @Override
+ protected void addDefaultIndexKey(Schema.Builder schemaBuilder) {
+ super.addDefaultIndexKey(schemaBuilder);
+
+ Schema currentSchema = schemaBuilder.build();
+ currentSchema.getPrimaryKey().ifPresent(pk -> schemaBuilder.index(pk.getColumnNames()));
+ }
+}
diff --git a/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/catalog/Flink23CatalogTest.java b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/catalog/Flink23CatalogTest.java
new file mode 100644
index 000000000..d1cc974d5
--- /dev/null
+++ b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/catalog/Flink23CatalogTest.java
@@ -0,0 +1,196 @@
+/*
+ * 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.fluss.flink.catalog;
+
+import org.apache.fluss.flink.FlinkConnectorOptions;
+import org.apache.fluss.flink.utils.FlinkConversions;
+import org.apache.fluss.metadata.TableDescriptor;
+import org.apache.fluss.metadata.TableInfo;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.testutils.common.MultiVersionTest;
+
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.catalog.CatalogMaterializedTable;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.DefaultIndex;
+import org.apache.flink.table.catalog.IntervalFreshness;
+import org.apache.flink.table.catalog.ResolvedCatalogMaterializedTable;
+import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.flink.table.catalog.StartMode;
+import org.apache.flink.table.catalog.UniqueConstraint;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.apache.fluss.record.TestData.DEFAULT_REMOTE_DATA_DIR;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link FlinkCatalog}. */
+public class Flink23CatalogTest extends FlinkCatalogTest {
+
+ protected ResolvedSchema createSchema() {
+ return new ResolvedSchema(
+ Arrays.asList(
+ Column.physical("first", DataTypes.STRING().notNull()),
+ Column.physical("second", DataTypes.INT()),
+ Column.physical("third", DataTypes.STRING().notNull())),
+ Collections.emptyList(),
+ UniqueConstraint.primaryKey("PK_first_third", Arrays.asList("first", "third")),
+ Collections.singletonList(
+ DefaultIndex.newIndex(
+ "INDEX_first_third", Arrays.asList("first", "third"))));
+ }
+
+ @Override
+ protected ResolvedCatalogMaterializedTable createResolvedCatalogMaterializedTable(
+ CatalogMaterializedTable origin,
+ ResolvedSchema resolvedSchema,
+ CatalogMaterializedTable.RefreshMode refreshMode,
+ IntervalFreshness intervalFreshness) {
+ return new ResolvedCatalogMaterializedTable(
+ origin,
+ resolvedSchema,
+ refreshMode,
+ intervalFreshness,
+ StartMode.of(StartMode.StartModeKind.FROM_BEGINNING));
+ }
+
+ /**
+ * Verifies that reading a legacy-shaped materialized-table payload (only {@code
+ * definition-query} persisted, no {@code original-query}/{@code expanded-query}) succeeds under
+ * Flink 2.3, where the underlying builder enforces non-null on the new fields. {@link
+ * FlinkConversions} must fall back to {@code definition-query} for both missing fields.
+ */
+ @Test
+ @MultiVersionTest
+ void testMaterializedTableFallsBackToDefinitionQueryForLegacyData() {
+ String definitionQuery = "SELECT order_id FROM t";
+
+ Map<String, String> customProperties = new HashMap<>();
+ customProperties.put(
+ FlinkConnectorOptions.MATERIALIZED_TABLE_DEFINITION_QUERY.key(), definitionQuery);
+ customProperties.put(
+ FlinkConnectorOptions.MATERIALIZED_TABLE_INTERVAL_FRESHNESS.key(), "5");
+ customProperties.put(
+ FlinkConnectorOptions.MATERIALIZED_TABLE_INTERVAL_FRESHNESS_TIME_UNIT.key(),
+ "SECOND");
+ customProperties.put(
+ FlinkConnectorOptions.MATERIALIZED_TABLE_LOGICAL_REFRESH_MODE.key(),
+ CatalogMaterializedTable.LogicalRefreshMode.CONTINUOUS.name());
+ customProperties.put(
+ FlinkConnectorOptions.MATERIALIZED_TABLE_REFRESH_MODE.key(),
+ CatalogMaterializedTable.RefreshMode.CONTINUOUS.name());
+ customProperties.put(
+ FlinkConnectorOptions.MATERIALIZED_TABLE_REFRESH_STATUS.key(),
+ CatalogMaterializedTable.RefreshStatus.INITIALIZING.name());
+ // Intentionally NO materialized-table.original-query / expanded-query entries —
+ // simulates a legacy persisted payload from before Flink 2.3.
+
+ org.apache.fluss.metadata.Schema flussSchema =
+ org.apache.fluss.metadata.Schema.newBuilder()
+ .column("order_id", org.apache.fluss.types.DataTypes.STRING())
+ .build();
+ TableDescriptor flussDescriptor =
+ TableDescriptor.builder()
+ .schema(flussSchema)
+ .distributedBy(1, Collections.singletonList("order_id"))
+ .customProperties(customProperties)
+ .build();
+
+ TablePath tablePath = TablePath.of("db", "table");
+ long currentMillis = System.currentTimeMillis();
+ TableInfo tableInfo =
+ TableInfo.of(
+ tablePath,
+ 1L,
+ 1,
+ flussDescriptor,
+ DEFAULT_REMOTE_DATA_DIR,
+ currentMillis,
+ currentMillis);
+
+ CatalogMaterializedTable flinkTable =
+ (CatalogMaterializedTable) FlinkConversions.toFlinkTable(tableInfo);
+
+ assertThat(flinkTable.getDefinitionQuery()).isEqualTo(definitionQuery);
+ // Flink 2.3 enforces non-null on original/expanded queries; verify the fallback
+ // actually populated these fields with the definition-query value.
+ assertThat(flinkTable.getOriginalQuery()).isEqualTo(definitionQuery);
+ assertThat(flinkTable.getExpandedQuery()).isEqualTo(definitionQuery);
+ }
+
+ @Test
+ @MultiVersionTest
+ void testMaterializedTableSerializesSeparateQueriesIntoCustomProperties() {
+ String originalQuery = "SELECT order_id, orig_ts FROM t";
+ String expandedQuery = "SELECT defalut.t.order_id, defalut.t.orig_ts FROM defalut.t";
+
+ ResolvedSchema resolvedSchema =
+ new ResolvedSchema(
+ Arrays.asList(
+ Column.physical(
+ "order_id",
+ org.apache.flink.table.api.DataTypes.STRING().notNull()),
+ Column.physical(
+ "orig_ts",
+ org.apache.flink.table.api.DataTypes.TIMESTAMP())),
+ Collections.emptyList(),
+ UniqueConstraint.primaryKey(
+ "PK_order_id", Collections.singletonList("order_id")),
+ Collections.emptyList());
+ Map<String, String> flinkOptions = new HashMap<>();
+ flinkOptions.put("k1", "v1");
+
+ CatalogMaterializedTable flinkMaterializedTable =
+ CatalogMaterializedTable.newBuilder()
+ .schema(Schema.newBuilder().fromResolvedSchema(resolvedSchema).build())
+ .comment("test comment")
+ .options(flinkOptions)
+ .originalQuery(originalQuery)
+ .expandedQuery(expandedQuery)
+ .freshness(IntervalFreshness.ofSecond(5))
+ .logicalRefreshMode(CatalogMaterializedTable.LogicalRefreshMode.CONTINUOUS)
+ .refreshMode(CatalogMaterializedTable.RefreshMode.CONTINUOUS)
+ .refreshStatus(CatalogMaterializedTable.RefreshStatus.INITIALIZING)
+ .build();
+
+ TableDescriptor flussTable =
+ FlinkConversions.toFlussTable(
+ new ResolvedCatalogMaterializedTable(
+ flinkMaterializedTable,
+ resolvedSchema,
+ CatalogMaterializedTable.RefreshMode.CONTINUOUS,
+ IntervalFreshness.ofSecond(5),
+ StartMode.of(StartMode.StartModeKind.FROM_BEGINNING)));
+
+ Map<String, String> customProperties = flussTable.getCustomProperties();
+
+ assertThat(customProperties)
+ .containsEntry(
+ FlinkConnectorOptions.MATERIALIZED_TABLE_ORIGINAL_QUERY.key(),
+ originalQuery);
+ assertThat(customProperties)
+ .containsEntry(
+ FlinkConnectorOptions.MATERIALIZED_TABLE_EXPANDED_QUERY.key(),
+ expandedQuery);
+ }
+}
diff --git a/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/catalog/Flink23MaterializedTableITCase.java b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/catalog/Flink23MaterializedTableITCase.java
new file mode 100644
index 000000000..ff62e8dc6
--- /dev/null
+++ b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/catalog/Flink23MaterializedTableITCase.java
@@ -0,0 +1,21 @@
+/*
+ * 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.fluss.flink.catalog;
+
+/** IT case for materialized table in Flink 2.3. */
+public class Flink23MaterializedTableITCase extends MaterializedTableITCase {}
diff --git a/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/catalog/Flink23TableFactoryTest.java b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/catalog/Flink23TableFactoryTest.java
new file mode 100644
index 000000000..a884745b2
--- /dev/null
+++ b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/catalog/Flink23TableFactoryTest.java
@@ -0,0 +1,29 @@
+/*
+ * 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.fluss.flink.catalog;
+
+import org.apache.flink.table.connector.source.LookupTableSource;
+import org.apache.flink.table.runtime.connector.source.LookupRuntimeProviderContext;
+
+/** Test for {@link FlinkTableFactory} in Flink 2.3. */
+public class Flink23TableFactoryTest extends FlinkTableFactoryTest {
+ @Override
+ protected LookupTableSource.LookupContext createLookupContext(int[][] lookupKeys) {
+ return new LookupRuntimeProviderContext(lookupKeys, false);
+ }
+}
diff --git a/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/metrics/Flink23MetricsITCase.java b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/metrics/Flink23MetricsITCase.java
new file mode 100644
index 000000000..26c8fcd85
--- /dev/null
+++ b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/metrics/Flink23MetricsITCase.java
@@ -0,0 +1,21 @@
+/*
+ * 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.fluss.flink.metrics;
+
+/** IT case for metrics in Flink 2.3. */
+public class Flink23MetricsITCase extends FlinkMetricsITCase {}
diff --git a/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/procedure/Flink23ProcedureITCase.java b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/procedure/Flink23ProcedureITCase.java
new file mode 100644
index 000000000..581f689ff
--- /dev/null
+++ b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/procedure/Flink23ProcedureITCase.java
@@ -0,0 +1,21 @@
+/*
+ * 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.fluss.flink.procedure;
+
+/** IT case for procedure in Flink 2.3. */
+public class Flink23ProcedureITCase extends FlinkProcedureITCase {}
diff --git a/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/security/acl/Flink23AuthorizationITCase.java b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/security/acl/Flink23AuthorizationITCase.java
new file mode 100644
index 000000000..8541a0d1a
--- /dev/null
+++ b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/security/acl/Flink23AuthorizationITCase.java
@@ -0,0 +1,21 @@
+/*
+ * 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.fluss.flink.security.acl;
+
+/** IT case for authorization in Flink 2.3. */
+public class Flink23AuthorizationITCase extends FlinkAuthorizationITCase {}
diff --git a/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/sink/Flink23ComplexTypeITCase.java b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/sink/Flink23ComplexTypeITCase.java
new file mode 100644
index 000000000..7521277b4
--- /dev/null
+++ b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/sink/Flink23ComplexTypeITCase.java
@@ -0,0 +1,21 @@
+/*
+ * 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.fluss.flink.sink;
+
+/** Integration tests for Array type support in Flink 2.3. */
+public class Flink23ComplexTypeITCase extends FlinkComplexTypeITCase {}
diff --git a/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/sink/Flink23TableSinkITCase.java b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/sink/Flink23TableSinkITCase.java
new file mode 100644
index 000000000..775f3689a
--- /dev/null
+++ b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/sink/Flink23TableSinkITCase.java
@@ -0,0 +1,34 @@
+/*
+ * 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.fluss.flink.sink;
+
+import org.apache.flink.table.api.config.ExecutionConfigOptions;
+
+/** IT case for {@link FlinkTableSink} in Flink 2.3. */
+public class Flink23TableSinkITCase extends FlinkTableSinkITCase {
+
+ /**
+ * Flink 2.3-specific override of the no-op hook defined in {@link FlinkTableSinkITCase}.
+ * Disables {@link ExecutionConfigOptions#TABLE_EXEC_SINK_REQUIRE_ON_CONFLICT} so that {@code
+ * FlinkChangelogModeInferenceProgram} does not reject plans whose inferred upsert key differs
+ * from the sink PK.
+ */
+ protected void disableSinkRequireOnConflict() {
+ tEnv.getConfig().set(ExecutionConfigOptions.TABLE_EXEC_SINK_REQUIRE_ON_CONFLICT, false);
+ }
+}
diff --git a/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/sink/Flink23UndoRecoveryITCase.java b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/sink/Flink23UndoRecoveryITCase.java
new file mode 100644
index 000000000..7b9b44705
--- /dev/null
+++ b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/sink/Flink23UndoRecoveryITCase.java
@@ -0,0 +1,21 @@
+/*
+ * 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.fluss.flink.sink;
+
+/** IT case for Undo Recovery functionality in Flink 2.3. */
+public class Flink23UndoRecoveryITCase extends UndoRecoveryITCase {}
diff --git a/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/source/Flink23BinlogVirtualTableITCase.java b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/source/Flink23BinlogVirtualTableITCase.java
new file mode 100644
index 000000000..e17e1affe
--- /dev/null
+++ b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/source/Flink23BinlogVirtualTableITCase.java
@@ -0,0 +1,21 @@
+/*
+ * 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.fluss.flink.source;
+
+/** IT case for {@link BinlogVirtualTableITCase} in Flink 2.3. */
+public class Flink23BinlogVirtualTableITCase extends BinlogVirtualTableITCase {}
diff --git a/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/source/Flink23ChangelogVirtualTableITCase.java b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/source/Flink23ChangelogVirtualTableITCase.java
new file mode 100644
index 000000000..ee27c922e
--- /dev/null
+++ b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/source/Flink23ChangelogVirtualTableITCase.java
@@ -0,0 +1,21 @@
+/*
+ * 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.fluss.flink.source;
+
+/** IT case for {@link ChangelogVirtualTableITCase} in Flink 2.3. */
+public class Flink23ChangelogVirtualTableITCase extends ChangelogVirtualTableITCase {}
diff --git a/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/source/Flink23DeltaJoinITCase.java b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/source/Flink23DeltaJoinITCase.java
new file mode 100644
index 000000000..b4f93193f
--- /dev/null
+++ b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/source/Flink23DeltaJoinITCase.java
@@ -0,0 +1,1101 @@
+/*
+ * 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.fluss.flink.source;
+
+import org.apache.fluss.config.ConfigOptions;
+import org.apache.fluss.flink.utils.FlinkTestBase;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.shaded.guava32.com.google.common.collect.ImmutableMap;
+import org.apache.fluss.testutils.common.MultiVersionTest;
+
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.api.config.ExecutionConfigOptions;
+import org.apache.flink.table.api.config.OptimizerConfigOptions;
+import org.apache.flink.types.Row;
+import org.apache.flink.util.CloseableIterator;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import javax.annotation.Nullable;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.apache.fluss.flink.FlinkConnectorOptions.BOOTSTRAP_SERVERS;
+import static org.apache.fluss.flink.source.testutils.FlinkRowAssertionsUtils.assertResultsIgnoreOrder;
+import static org.apache.fluss.server.testutils.FlussClusterExtension.BUILTIN_DATABASE;
+import static org.apache.fluss.testutils.DataTestUtils.row;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** IT case for Delta Join optimization in Flink 2.3. */
+@MultiVersionTest
+public class Flink23DeltaJoinITCase extends FlinkTestBase {
+
+ private static final String CATALOG_NAME = "test_catalog";
+
+ private StreamTableEnvironment tEnv;
+
+ @BeforeEach
+ public void beforeEach() {
+ bootstrapServers = conn.getConfiguration().get(ConfigOptions.BOOTSTRAP_SERVERS).get(0);
+
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ tEnv = StreamTableEnvironment.create(env, EnvironmentSettings.inStreamingMode());
+
+ tEnv.executeSql(
+ String.format(
+ "create catalog %s with ('type' = 'fluss', '%s' = '%s')",
+ CATALOG_NAME, BOOTSTRAP_SERVERS.key(), bootstrapServers));
+ tEnv.useCatalog(CATALOG_NAME);
+ tEnv.executeSql(String.format("create database if not exists `%s`", DEFAULT_DB));
+ tEnv.useDatabase(DEFAULT_DB);
+
+ // start two jobs for this test: one for DML involving the delta join, and the other for DQL
+ // to query the results of the sink table
+ tEnv.getConfig().set(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM, 2);
+ // Set FORCE strategy for delta join
+ tEnv.getConfig()
+ .set(
+ OptimizerConfigOptions.TABLE_OPTIMIZER_DELTA_JOIN_STRATEGY,
+ OptimizerConfigOptions.DeltaJoinStrategy.FORCE);
+ }
+
+ /**
+ * Disables {@link ExecutionConfigOptions#TABLE_EXEC_SINK_REQUIRE_ON_CONFLICT} so the {@code
+ * FlinkChangelogModeInferenceProgram} does not reject plans whose inferred upsert key differs
+ * from the sink PK before the {@code StreamPhysicalDeltaJoinForceValidator} has a chance to
+ * run.
+ */
+ private void disableSinkRequireOnConflict() {
+ tEnv.getConfig().set(ExecutionConfigOptions.TABLE_EXEC_SINK_REQUIRE_ON_CONFLICT, false);
+ }
+
+ @AfterEach
+ void after() {
+ tEnv.useDatabase(BUILTIN_DATABASE);
+ tEnv.executeSql(String.format("drop database `%s` cascade", DEFAULT_DB));
+ }
+
+ /**
+ * Creates a source table with specified schema and options.
+ *
+ * @param tableName the name of the table
+ * @param columns column definitions (e.g., "a1 int, b1 varchar, c1 bigint, d1 int, e1 bigint")
+ * @param primaryKey primary key columns (e.g., "c1, d1")
+ * @param bucketKey bucket key column (e.g., "c1")
+ * @param extraOptions additional WITH options (e.g., "lookup.cache" = "partial"), or null
+ */
+ private void createSource(
+ String tableName,
+ String columns,
+ String primaryKey,
+ String bucketKey,
+ @Nullable String partitionKey,
+ @Nullable Map<String, String> extraOptions) {
+ Map<String, String> withOptions = new HashMap<>();
+ if (extraOptions != null) {
+ withOptions.putAll(extraOptions);
+ }
+ withOptions.put("connector", "fluss");
+ withOptions.put("bucket.key", bucketKey);
+ StringBuilder ddlBuilder = new StringBuilder();
+ ddlBuilder.append(
+ String.format(
+ "create table %s ( %s, primary key (%s) NOT ENFORCED )",
+ tableName, columns, primaryKey));
+ if (partitionKey != null) {
+ ddlBuilder.append(String.format(" partitioned by (%s)", partitionKey));
+ withOptions.put("table.auto-partition.enabled", "true");
+ withOptions.put("table.auto-partition.time-unit", "year");
+ }
+ ddlBuilder.append(
+ String.format(
+ " with (%s)",
+ withOptions.entrySet().stream()
+ .map(e -> String.format("'%s' = '%s'", e.getKey(), e.getValue()))
+ .collect(Collectors.joining(", "))));
+
+ tEnv.executeSql(ddlBuilder.toString());
+ }
+
+ /**
+ * Creates a sink table with specified columns.
+ *
+ * @param tableName the name of the table
+ * @param columns the column definitions (e.g., "a1 int, c1 bigint, a2 int")
+ * @param primaryKey the primary key columns (e.g., "c1, d1, c2, d2")
+ */
+ private void createSink(String tableName, String columns, String primaryKey) {
+ tEnv.executeSql(
+ String.format(
+ "create table %s ("
+ + " %s, "
+ + " primary key (%s) NOT ENFORCED"
+ + ") with ("
+ + " 'connector' = 'fluss'"
+ + ")",
+ tableName, columns, primaryKey));
+ }
+
+ @Test
+ void testDeltaJoin() throws Exception {
+ // disable cache to get stable results with updating
+ tEnv.getConfig().set(ExecutionConfigOptions.TABLE_EXEC_DELTA_JOIN_CACHE_ENABLED, false);
+ String leftTableName = "left_table";
+ String rightTableName = "right_table";
+ String sinkTableName = "sink_table";
+
+ createSource(
+ leftTableName,
+ "a1 int, b1 varchar, c1 bigint, d1 int, e1 bigint",
+ "c1, d1",
+ "c1",
+ null,
+ ImmutableMap.of("table.delete.behavior", "IGNORE"));
+ createSource(
+ rightTableName,
+ "a2 int, b2 varchar, c2 bigint, d2 int, e2 bigint",
+ "c2, d2",
+ "c2",
+ null,
+ ImmutableMap.of("table.delete.behavior", "IGNORE"));
+ createSink(
+ sinkTableName,
+ "a1 int, b1 varchar, c1 bigint, d1 int, e1 bigint, a2 int, b2 varchar, c2 bigint, d2 int, e2 bigint",
+ "c1, d1, c2, d2");
+
+ List<InternalRow> rows1 =
+ Arrays.asList(
+ row(1, "v1", 100L, 1, 10000L),
+ row(2, "v2", 100L, 2, 20000L),
+ row(3, "v3", 300L, 3, 30000L),
+ row(4, "v4", 400L, 4, 40000L),
+ // update
+ row(5, "v5", 100L, 1, 50000L));
+ TablePath leftTablePath = TablePath.of(DEFAULT_DB, leftTableName);
+ writeRows(conn, leftTablePath, rows1, false);
+ // wait for the first snapshot to finish to get the stable result
+ FLUSS_CLUSTER_EXTENSION.triggerAndWaitSnapshot(leftTablePath);
+
+ List<InternalRow> rows2 =
+ Arrays.asList(
+ row(1, "v1", 100L, 1, 10000L),
+ row(2, "v2", 200L, 2, 20000L),
+ row(3, "v3", 330L, 4, 30000L),
+ row(4, "v4", 500L, 4, 50000L),
+ // update
+ row(6, "v6", 100L, 1, 60000L));
+ TablePath rightTablePath = TablePath.of(DEFAULT_DB, rightTableName);
+ writeRows(conn, rightTablePath, rows2, false);
+ // wait for the first snapshot to finish to get the stable result
+ FLUSS_CLUSTER_EXTENSION.triggerAndWaitSnapshot(rightTablePath);
+
+ String sql =
+ String.format(
+ "INSERT INTO %s SELECT * FROM %s INNER JOIN %s ON c1 = c2",
+ sinkTableName, leftTableName, rightTableName);
+
+ assertThat(tEnv.explainSql(sql))
+ .contains("DeltaJoin(joinType=[InnerJoin], where=[(c1 = c2)]");
+
+ tEnv.executeSql(sql);
+
+ CloseableIterator<Row> collected =
+ tEnv.executeSql(String.format("select * from %s", sinkTableName)).collect();
+ List<String> expected =
+ Arrays.asList(
+ "+I[5, v5, 100, 1, 50000, 6, v6, 100, 1, 60000]",
+ "-U[5, v5, 100, 1, 50000, 6, v6, 100, 1, 60000]",
+ "+U[5, v5, 100, 1, 50000, 6, v6, 100, 1, 60000]",
+ "+I[2, v2, 100, 2, 20000, 6, v6, 100, 1, 60000]",
+ "-U[2, v2, 100, 2, 20000, 6, v6, 100, 1, 60000]",
+ "+U[2, v2, 100, 2, 20000, 6, v6, 100, 1, 60000]");
+ assertResultsIgnoreOrder(collected, expected, true);
+ }
+
+ @Test
+ void testDeltaJoinOnPrimaryKey() throws Exception {
+ // disable cache to get stable results with updating
+ tEnv.getConfig().set(ExecutionConfigOptions.TABLE_EXEC_DELTA_JOIN_CACHE_ENABLED, false);
+ String leftTableName = "left_table";
+ String rightTableName = "right_table";
+ String sinkTableName = "sink_table";
+
+ createSource(
+ leftTableName,
+ "a1 int, b1 varchar, c1 bigint, d1 int, e1 bigint",
+ "c1, d1",
+ "c1",
+ null,
+ ImmutableMap.of("table.delete.behavior", "IGNORE"));
+ createSource(
+ rightTableName,
+ "a2 int, b2 varchar, c2 bigint, d2 int, e2 bigint",
+ "c2, d2",
+ "c2",
+ null,
+ ImmutableMap.of("table.delete.behavior", "IGNORE"));
+ createSink(
+ sinkTableName,
+ "a1 int, b1 varchar, c1 bigint, d1 int, e1 bigint, a2 int, b2 varchar, c2 bigint, d2 int, e2 bigint",
+ "c1, d1, c2, d2");
+
+ List<InternalRow> rows1 =
+ Arrays.asList(
+ row(1, "v1", 100L, 1, 10000L),
+ row(2, "v2", 200L, 2, 20000L),
+ row(3, "v3", 300L, 3, 30000L),
+ row(4, "v4", 400L, 4, 40000L),
+ // update
+ row(5, "v5", 100L, 1, 50000L));
+ TablePath leftTablePath = TablePath.of(DEFAULT_DB, leftTableName);
+ writeRows(conn, leftTablePath, rows1, false);
+ // wait for the first snapshot to finish to get the stable result
+ FLUSS_CLUSTER_EXTENSION.triggerAndWaitSnapshot(leftTablePath);
+
+ List<InternalRow> rows2 =
+ Arrays.asList(
+ row(1, "v1", 100L, 1, 10000L),
+ row(2, "v2", 200L, 2, 20000L),
+ row(3, "v3", 300L, 4, 30000L),
+ row(4, "v4", 500L, 4, 50000L),
+ // update
+ row(6, "v6", 100L, 1, 60000L));
+ TablePath rightTablePath = TablePath.of(DEFAULT_DB, rightTableName);
+ writeRows(conn, rightTablePath, rows2, false);
+ // wait for the first snapshot to finish to get the stable result
+ FLUSS_CLUSTER_EXTENSION.triggerAndWaitSnapshot(rightTablePath);
+
+ String sql =
+ String.format(
+ "INSERT INTO %s SELECT * FROM %s INNER JOIN %s ON c1 = c2 AND d1 = d2",
+ sinkTableName, leftTableName, rightTableName);
+
+ assertThat(tEnv.explainSql(sql))
+ .contains("DeltaJoin(joinType=[InnerJoin], where=[((c1 = c2) AND (d1 = d2))]");
+
+ tEnv.executeSql(sql);
+
+ CloseableIterator<Row> collected =
+ tEnv.executeSql(String.format("select * from %s", sinkTableName)).collect();
+ List<String> expected =
+ Arrays.asList(
+ "+I[5, v5, 100, 1, 50000, 6, v6, 100, 1, 60000]",
+ "-U[5, v5, 100, 1, 50000, 6, v6, 100, 1, 60000]",
+ "+U[5, v5, 100, 1, 50000, 6, v6, 100, 1, 60000]",
+ "+I[2, v2, 200, 2, 20000, 2, v2, 200, 2, 20000]",
+ "-U[2, v2, 200, 2, 20000, 2, v2, 200, 2, 20000]",
+ "+U[2, v2, 200, 2, 20000, 2, v2, 200, 2, 20000]");
+ assertResultsIgnoreOrder(collected, expected, true);
+ }
+
+ @Test
+ void testDeltaJoinWithCalc() throws Exception {
+ String leftTableName = "left_table_proj";
+ createSource(
+ leftTableName,
+ "a1 int, b1 varchar, c1 bigint, d1 int, e1 bigint",
+ "c1, d1",
+ "c1",
+ null,
+ ImmutableMap.of("table.delete.behavior", "IGNORE"));
+
+ List<InternalRow> rows1 =
+ Arrays.asList(
+ row(1, "v1", 100L, 1, 10000L),
+ row(2, "v2", 200L, 2, 20000L),
+ row(3, "v1", 300L, 3, 30000L));
+ TablePath leftTablePath = TablePath.of(DEFAULT_DB, leftTableName);
+ writeRows(conn, leftTablePath, rows1, false);
+
+ String rightTableName = "right_table_proj";
+ createSource(
+ rightTableName,
+ "a2 int, b2 varchar, c2 bigint, d2 int, e2 bigint",
+ "c2",
+ "c2",
+ null,
+ ImmutableMap.of("table.delete.behavior", "IGNORE"));
+
+ List<InternalRow> rows2 =
+ Arrays.asList(
+ row(1, "v1", 100L, 1, 10000L),
+ row(2, "v2", 200L, 2, 20000L),
+ row(3, "v3", 300L, 4, 30000L));
+ TablePath rightTablePath = TablePath.of(DEFAULT_DB, rightTableName);
+ writeRows(conn, rightTablePath, rows2, false);
+
+ String sinkTableName = "sink_table_proj";
+ createSink(sinkTableName, "a1 int, c1 bigint, d1 int, a2 int", "c1, d1");
+
+ String sql =
+ String.format(
+ "INSERT INTO %s SELECT a1, c1, d1, a2 FROM ("
+ + " SELECT * FROM %s WHERE d1 > 1"
+ + ") INNER JOIN ("
+ + " SELECT * FROM %s WHERE c2 < 300"
+ + ") ON c1 = c2",
+ sinkTableName, leftTableName, rightTableName);
+
+ assertThat(tEnv.explainSql(sql))
+ .contains("DeltaJoin(joinType=[InnerJoin], where=[(c1 = c2)]");
+
+ tEnv.executeSql(sql);
+
+ CloseableIterator<Row> collected =
+ tEnv.executeSql(String.format("select * from %s", sinkTableName)).collect();
+ List<String> expected =
+ Arrays.asList("+I[2, 200, 2, 2]", "-U[2, 200, 2, 2]", "+U[2, 200, 2, 2]");
+ assertResultsIgnoreOrder(collected, expected, true);
+ }
+
+ @Test
+ void testDeltaJoinWithAppendOnlySourceAndCalc() throws Exception {
+ String leftTableName = "left_table_proj";
+ createSource(
+ leftTableName,
+ "a1 int, b1 varchar, c1 bigint, d1 int, e1 bigint",
+ "c1, d1",
+ "c1",
+ null,
+ ImmutableMap.of("table.merge-engine", "first_row"));
+
+ List<InternalRow> rows1 =
+ Arrays.asList(
+ row(1, "v1", 100L, 1, 10000L),
+ row(2, "v2", 200L, 2, 20000L),
+ row(3, "v1", 300L, 3, 30000L));
+ TablePath leftTablePath = TablePath.of(DEFAULT_DB, leftTableName);
+ writeRows(conn, leftTablePath, rows1, false);
+
+ String rightTableName = "right_table_proj";
+ createSource(
+ rightTableName,
+ "a2 int, b2 varchar, c2 bigint, d2 int, e2 bigint",
+ "c2, d2",
+ "c2",
+ null,
+ ImmutableMap.of("table.merge-engine", "first_row"));
+
+ List<InternalRow> rows2 =
+ Arrays.asList(
+ row(1, "v1", 100L, 1, 10000L),
+ row(2, "v3", 200L, 2, 20000L),
+ row(3, "v4", 400L, 4, 30000L));
+ TablePath rightTablePath = TablePath.of(DEFAULT_DB, rightTableName);
+ writeRows(conn, rightTablePath, rows2, false);
+
+ String sinkTableName = "sink_table_proj";
+ createSink(sinkTableName, "a1 int, c1 bigint, a2 int", "c1");
+
+ String sql =
+ String.format(
+ "INSERT INTO %s SELECT a1, c1, a2 FROM %s INNER JOIN %s ON c1 = c2 WHERE a1 > 1",
+ sinkTableName, leftTableName, rightTableName);
+
+ assertThat(tEnv.explainSql(sql))
+ .contains("DeltaJoin(joinType=[InnerJoin], where=[(c1 = c2)]");
+
+ tEnv.executeSql(sql);
+
+ CloseableIterator<Row> collected =
+ tEnv.executeSql(String.format("select * from %s", sinkTableName)).collect();
+ List<String> expected = Arrays.asList("+I[2, 200, 2]", "-U[2, 200, 2]", "+U[2, 200, 2]");
+ assertResultsIgnoreOrder(collected, expected, true);
+ }
+
+ @Test
+ void testDeltaJoinFailsWhenFilterOnNonUpsertKeys() {
+ String leftTableName = "left_table_force_fail";
+ createSource(
+ leftTableName,
+ "a1 int, b1 varchar, c1 bigint, d1 int, e1 bigint",
+ "c1, d1",
+ "c1",
+ null,
+ ImmutableMap.of("table.delete.behavior", "IGNORE"));
+
+ String rightTableName = "right_table_force_fail";
+ createSource(
+ rightTableName,
+ "a2 int, b2 varchar, c2 bigint, d2 int, e2 bigint",
+ "c2, d2",
+ "c2",
+ null,
+ ImmutableMap.of("table.delete.behavior", "IGNORE"));
+
+ String sinkTableName = "sink_table_force_fail";
+ createSink(
+ sinkTableName,
+ "a1 int, b1 varchar, c1 bigint, d1 int, e1 bigint, a2 int, b2 varchar, c2 bigint, d2 int, e2 bigint",
+ "c1, d1, c2, d2");
+
+ // Filter on e1 > e2, where e1 and e2 are NOT part of the upsert key
+ // TODO we can add a UpsertFilterOperator that can convert the un-match-filter UPSERT record
+ // into DELETE record.
+ String sql =
+ String.format(
+ "INSERT INTO %s SELECT * FROM %s INNER JOIN %s ON c1 = c2 AND d1 = d2 WHERE e1 > e2",
+ sinkTableName, leftTableName, rightTableName);
+
+ assertThatThrownBy(() -> tEnv.explainSql(sql))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("doesn't support to do delta join optimization");
+
+ // Non-equiv-cond on e1 > e2, where e1 and e2 are NOT part of the upsert key
+ String sql2 =
+ String.format(
+ "INSERT INTO %s SELECT * FROM %s INNER JOIN %s ON c1 = c2 AND d1 = d2 AND e1 > e2",
+ sinkTableName, leftTableName, rightTableName);
+
+ assertThatThrownBy(() -> tEnv.explainSql(sql2))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("doesn't support to do delta join optimization");
+ }
+
+ @Test
+ void testDeltaJoinWithNonEquiConditionOnUpsertKeys() throws Exception {
+ String leftTableName = "left_table_nonequi_upsert";
+ createSource(
+ leftTableName,
+ "a1 int, b1 varchar, c1 bigint, d1 int, e1 bigint",
+ "c1, d1, e1",
+ "c1, d1",
+ null,
+ ImmutableMap.of("table.delete.behavior", "IGNORE"));
+
+ List<InternalRow> rows1 =
+ Arrays.asList(
+ row(1, "v1", 100L, 1, 10000L),
+ row(2, "v2", 200L, 2, 20001L),
+ row(3, "v3", 300L, 3, 30000L),
+ // Add row with same PK (100, 1) to generate UPDATE in CDC mode
+ // This row is filtered out by (e2 / 100) <> c2, so doesn't affect join
+ // result
+ row(4, "v1_updated", 100L, 1, 10000L));
+ TablePath leftTablePath = TablePath.of(DEFAULT_DB, leftTableName);
+ writeRows(conn, leftTablePath, rows1, false);
+
+ String rightTableName = "right_table_nonequi_upsert";
+ createSource(
+ rightTableName,
+ "a2 int, b2 varchar, c2 bigint, d2 int, e2 bigint",
+ "c2, d2",
+ "c2",
+ null,
+ ImmutableMap.of("table.delete.behavior", "IGNORE"));
+
+ List<InternalRow> rows2 =
+ Arrays.asList(
+ row(1, "v1", 100L, 1, 10000L),
+ row(2, "v4", 200L, 2, 20000L),
+ row(3, "v5", 300L, 4, 40000L),
+ // Add row with same PK (100, 1) to generate UPDATE in CDC mode
+ // This row is filtered out by (e2 / 100) <> c2, so doesn't affect join
+ // result
+ row(5, "v1_updated", 100L, 1, 10000L));
+ TablePath rightTablePath = TablePath.of(DEFAULT_DB, rightTableName);
+ writeRows(conn, rightTablePath, rows2, false);
+
+ String sinkTableName = "sink_table_nonequi_upsert";
+ createSink(sinkTableName, "a1 int, c1 bigint, d1 int, e1 bigint, a2 int", "c1, d1, e1");
+
+ String sql =
+ String.format(
+ "INSERT INTO %s SELECT a1, c1, d1, e1, a2 FROM %s INNER JOIN %s "
+ + "ON c1 = c2 AND d1 = d2 AND e1 <> (c2 * 100)",
+ sinkTableName, leftTableName, rightTableName);
+
+ assertThat(tEnv.explainSql(sql))
+ .contains(
+ "DeltaJoin(joinType=[InnerJoin], where=[((c1 = c2) AND (d1 = d2) AND (e1 <> (c2 * 100)))]");
+
+ tEnv.executeSql(sql);
+
+ CloseableIterator<Row> collected =
+ tEnv.executeSql(String.format("select * from %s", sinkTableName)).collect();
+ List<String> expected =
+ Arrays.asList(
+ "+I[2, 200, 2, 20001, 2]",
+ "-U[2, 200, 2, 20001, 2]",
+ "+U[2, 200, 2, 20001, 2]");
+ assertResultsIgnoreOrder(collected, expected, true);
+ }
+
+ @Test
+ void testDeltaJoinWithAppendOnlySourceAndNonEquiCondition() throws Exception {
+ String leftTableName = "left_table_nonequi_insert";
+ createSource(
+ leftTableName,
+ "a1 int, b1 varchar, c1 bigint, d1 int, e1 bigint",
+ "c1, d1",
+ "c1",
+ null,
+ ImmutableMap.of("table.merge-engine", "first_row"));
+
+ List<InternalRow> rows1 =
+ Arrays.asList(
+ row(1, "v1", 100L, 1, 10000L),
+ row(2, "v2", 200L, 2, 20000L),
+ row(3, "v3", 300L, 3, 5000L));
+ TablePath leftTablePath = TablePath.of(DEFAULT_DB, leftTableName);
+ writeRows(conn, leftTablePath, rows1, false);
+
+ String rightTableName = "right_table_nonequi_insert";
+ createSource(
+ rightTableName,
+ "a2 int, b2 varchar, c2 bigint, d2 int, e2 bigint",
+ "c2, d2",
+ "c2",
+ null,
+ ImmutableMap.of("table.merge-engine", "first_row"));
+
+ List<InternalRow> rows2 =
+ Arrays.asList(
+ row(1, "v1", 100L, 1, 8000L),
+ row(2, "v4", 200L, 2, 15000L),
+ row(3, "v5", 300L, 3, 3000L));
+ TablePath rightTablePath = TablePath.of(DEFAULT_DB, rightTableName);
+ writeRows(conn, rightTablePath, rows2, false);
+
+ String sinkTableName = "sink_table_nonequi_insert";
+ createSink(
+ sinkTableName, "a1 int, c1 bigint, d1 int, e1 bigint, a2 int, e2 bigint", "c1, d1");
+
+ // INSERT_ONLY sources with non-equi condition on non-upsert key fields (e1, e2)
+ // This should succeed because INSERT_ONLY mode allows any non-equi conditions
+ String sql =
+ String.format(
+ "INSERT INTO %s SELECT a1, c1, d1, e1, a2, e2 FROM %s INNER JOIN %s ON c1 = c2 AND d1 = d2 AND e1 > e2",
+ sinkTableName, leftTableName, rightTableName);
+
+ assertThat(tEnv.explainSql(sql))
+ .contains(
+ "DeltaJoin(joinType=[InnerJoin], where=[((c1 = c2) AND (d1 = d2) AND (e1 > e2))]");
+
+ tEnv.executeSql(sql);
+
+ CloseableIterator<Row> collected =
+ tEnv.executeSql(String.format("select * from %s", sinkTableName)).collect();
+ // Rows where e1 > e2:
+ // Row 1: e1=10000 > e2=8000 ✓
+ // Row 2: e1=20000 > e2=15000 ✓
+ // Row 3: e1=5000 > e2=3000 ✓
+ List<String> expected =
+ Arrays.asList(
+ "+I[1, 100, 1, 10000, 1, 8000]",
+ "-U[1, 100, 1, 10000, 1, 8000]",
+ "+U[1, 100, 1, 10000, 1, 8000]",
+ "+I[2, 200, 2, 20000, 2, 15000]",
+ "-U[2, 200, 2, 20000, 2, 15000]",
+ "+U[2, 200, 2, 20000, 2, 15000]",
+ "+I[3, 300, 3, 5000, 3, 3000]",
+ "-U[3, 300, 3, 5000, 3, 3000]",
+ "+U[3, 300, 3, 5000, 3, 3000]");
+ assertResultsIgnoreOrder(collected, expected, true);
+ }
+
+ @Test
+ void testDeltaJoinWithLookupCache() throws Exception {
+ String leftTableName = "left_table_cache";
+ createSource(
+ leftTableName,
+ "a1 int, c1 bigint, d1 int",
+ "c1, d1",
+ "c1",
+ null,
+ ImmutableMap.of("table.delete.behavior", "IGNORE"));
+ List<InternalRow> rows1 = Collections.singletonList(row(1, 100L, 1));
+ writeRows(conn, TablePath.of(DEFAULT_DB, leftTableName), rows1, false);
+
+ String rightTableName = "right_table_cache";
+ createSource(
+ rightTableName,
+ "a2 int, c2 bigint, d2 int",
+ "c2",
+ "c2",
+ null,
+ ImmutableMap.of(
+ "table.delete.behavior",
+ "IGNORE",
+ "lookup.cache",
+ "partial",
+ "lookup.partial-cache.max-rows",
+ "100"));
+ List<InternalRow> rows2 = Collections.singletonList(row(1, 100L, 1));
+ writeRows(conn, TablePath.of(DEFAULT_DB, rightTableName), rows2, false);
+
+ String sinkTableName = "sink_table_cache";
+ createSink(sinkTableName, "a1 int, c1 bigint, d1 int, a2 int", "c1, d1");
+
+ String sql =
+ String.format(
+ "INSERT INTO %s SELECT a1, c1, d1, a2 FROM %s AS T1 INNER JOIN %s AS T2 ON T1.c1 = T2.c2",
+ sinkTableName, leftTableName, rightTableName);
+
+ assertThat(tEnv.explainSql(sql))
+ .contains("DeltaJoin(joinType=[InnerJoin], where=[(c1 = c2)]");
+
+ tEnv.executeSql(sql);
+
+ CloseableIterator<Row> collected =
+ tEnv.executeSql(String.format("select * from %s", sinkTableName)).collect();
+ List<String> expected =
+ Arrays.asList("+I[1, 100, 1, 1]", "-U[1, 100, 1, 1]", "+U[1, 100, 1, 1]");
+ assertResultsIgnoreOrder(collected, expected, true);
+ }
+
+ @Test
+ void testDeltaJoinWithPartitionedTable() throws Exception {
+ String leftTableName = "left_table_partitioned";
+ createSource(
+ leftTableName,
+ "a1 int, b1 varchar, c1 bigint, d1 int, pt1 varchar",
+ "c1, d1, pt1",
+ "c1",
+ "pt1",
+ ImmutableMap.of("table.delete.behavior", "IGNORE"));
+ TablePath leftTablePath = TablePath.of(DEFAULT_DB, leftTableName);
+ Iterator<String> leftPartitionIterator =
+ waitUntilPartitions(FLUSS_CLUSTER_EXTENSION.getZooKeeperClient(), leftTablePath)
+ .values()
+ .iterator();
+ // pick two partition to insert data
+ String leftPartition1 = leftPartitionIterator.next();
+ String leftPartition2 = leftPartitionIterator.next();
+ List<InternalRow> rows1 =
+ Arrays.asList(
+ row(1, "v1", 100L, 1000, leftPartition1),
+ row(2, "v2", 200L, 2000, leftPartition1),
+ row(3, "v3", 100L, 3000, leftPartition2),
+ row(4, "v4", 400L, 4000, leftPartition2));
+ writeRows(conn, leftTablePath, rows1, false);
+
+ String rightTableName = "right_table_partitioned";
+ createSource(
+ rightTableName,
+ "a2 int, b2 varchar, c2 bigint, d2 int, pt2 varchar",
+ "c2, pt2",
+ "c2",
+ "pt2",
+ ImmutableMap.of("table.delete.behavior", "IGNORE"));
+ TablePath rightTablePath = TablePath.of(DEFAULT_DB, rightTableName);
+ Iterator<String> rightPartitionIterator =
+ waitUntilPartitions(FLUSS_CLUSTER_EXTENSION.getZooKeeperClient(), rightTablePath)
+ .values()
+ .iterator();
+ // pick two partition to insert data
+ String rightPartition1 = rightPartitionIterator.next();
+ String rightPartition2 = rightPartitionIterator.next();
+ List<InternalRow> rows2 =
+ Arrays.asList(
+ row(1, "v1", 100L, 1000, rightPartition1),
+ row(4, "v4", 400L, 3000, rightPartition2));
+ writeRows(conn, rightTablePath, rows2, false);
+
+ String sinkTableName = "sink_table";
+ createSink(
+ sinkTableName,
+ "a1 int, b1 varchar, c1 bigint, d1 int, pt1 varchar, b2 varchar",
+ "c1, d1, pt1");
+
+ String sql =
+ String.format(
+ "INSERT INTO %s SELECT a1, b1, c1, d1, pt1, b2 FROM %s AS T1 INNER JOIN %s AS T2 "
+ + "ON T1.c1 = T2.c2 AND T1.pt1 = T2.pt2",
+ sinkTableName, leftTableName, rightTableName);
+
+ assertThat(tEnv.explainSql(sql))
+ .contains("DeltaJoin(joinType=[InnerJoin], where=[((c1 = c2) AND (pt1 = pt2))]");
+
+ tEnv.executeSql(sql);
+
+ CloseableIterator<Row> collected =
+ tEnv.executeSql(String.format("select * from %s", sinkTableName)).collect();
+ List<String> expected =
+ Arrays.asList(
+ String.format("+I[1, v1, 100, 1000, %s, v1]", leftPartition1),
+ String.format("-U[1, v1, 100, 1000, %s, v1]", leftPartition1),
+ String.format("+U[1, v1, 100, 1000, %s, v1]", leftPartition1),
+ String.format("+I[4, v4, 400, 4000, %s, v4]", leftPartition2),
+ String.format("-U[4, v4, 400, 4000, %s, v4]", leftPartition2),
+ String.format("+U[4, v4, 400, 4000, %s, v4]", leftPartition2));
+ assertResultsIgnoreOrder(collected, expected, true);
+ }
+
+ @Test
+ void testDeltaJoinFailsWhenSourceHasDelete() {
+ String leftTableName = "left_table_delete_force";
+ createSource(
+ leftTableName,
+ "a1 int, b1 varchar, c1 bigint, d1 int, e1 bigint",
+ "c1, d1",
+ "c1",
+ null,
+ null);
+
+ String rightTableName = "right_table_delete_force";
+ createSource(
+ rightTableName,
+ "a2 int, b2 varchar, c2 bigint, d2 int, e2 bigint",
+ "c2, d2",
+ "c2",
+ null,
+ null);
+
+ String sinkTableName = "sink_table_delete_force";
+ createSink(
+ sinkTableName,
+ "a1 int, b1 varchar, c1 bigint, d1 int, e1 bigint, a2 int, b2 varchar, c2 bigint, d2 int, e2 bigint",
+ "c1, d1, c2, d2");
+
+ String sql =
+ String.format(
+ "INSERT INTO %s SELECT * FROM %s INNER JOIN %s ON c1 = c2 AND d1 = d2",
+ sinkTableName, leftTableName, rightTableName);
+
+ assertThatThrownBy(() -> tEnv.explainSql(sql))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("doesn't support to do delta join optimization");
+ }
+
+ @Test
+ void testDeltaJoinWithJoinKeyExceedsPrimaryKey() {
+ disableSinkRequireOnConflict();
+ String leftTableName = "left_table_exceed_pk";
+ createSource(
+ leftTableName,
+ "a1 int, b1 varchar, c1 bigint, d1 int, e1 bigint",
+ "c1, d1",
+ "c1",
+ null,
+ ImmutableMap.of("table.delete.behavior", "IGNORE"));
+
+ String rightTableName = "right_table_exceed_pk";
+ createSource(
+ rightTableName,
+ "a2 int, b2 varchar, c2 bigint, d2 int, e2 bigint",
+ "c2, d2",
+ "c2",
+ null,
+ ImmutableMap.of("table.delete.behavior", "IGNORE"));
+
+ String sinkTableName = "sink_table_exceed_pk";
+ createSink(
+ sinkTableName, "a1 int, c1 bigint, d1 int, e1 bigint, a2 int, e2 bigint", "c1, d1");
+
+ String sql =
+ String.format(
+ "INSERT INTO %s SELECT a1, c1, d1, e1, a2, e2 FROM %s INNER JOIN %s ON c1 = c2 AND d1 = d2 AND e1 = e2",
+ sinkTableName, leftTableName, rightTableName);
+
+ assertThatThrownBy(() -> tEnv.explainSql(sql))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("doesn't support to do delta join optimization");
+ }
+
+ @Test
+ void testDeltaJoinWithAppendOnlySourceAndJoinKeyExceedsPrimaryKey() throws Exception {
+ String leftTableName = "left_table_exceed_pk";
+ createSource(
+ leftTableName,
+ "a1 int, b1 varchar, c1 bigint, d1 int, e1 bigint",
+ "c1, d1",
+ "c1",
+ null,
+ ImmutableMap.of("table.merge-engine", "first_row"));
+
+ List<InternalRow> rows1 =
+ Arrays.asList(row(1, "v1", 100L, 1, 10000L), row(2, "v2", 200L, 2, 20000L));
+ TablePath leftTablePath = TablePath.of(DEFAULT_DB, leftTableName);
+ writeRows(conn, leftTablePath, rows1, false);
+
+ String rightTableName = "right_table_exceed_pk";
+ createSource(
+ rightTableName,
+ "a2 int, b2 varchar, c2 bigint, d2 int, e2 bigint",
+ "c2, d2",
+ "c2",
+ null,
+ ImmutableMap.of("table.merge-engine", "first_row"));
+
+ List<InternalRow> rows2 =
+ Arrays.asList(row(1, "v1", 100L, 1, 10000L), row(2, "v3", 200L, 2, 99999L));
+ TablePath rightTablePath = TablePath.of(DEFAULT_DB, rightTableName);
+ writeRows(conn, rightTablePath, rows2, false);
+
+ String sinkTableName = "sink_table_exceed_pk";
+ createSink(
+ sinkTableName, "a1 int, c1 bigint, d1 int, e1 bigint, a2 int, e2 bigint", "c1, d1");
+
+ // Join on PK (c1, d1) + additional non-PK field (e1)
+ // This should succeed because join keys {c1, d1, e1} contain complete PK {c1, d1}
+ // The e1 = e2 condition is applied as a post-lookup equi-condition filter
+ String sql =
+ String.format(
+ "INSERT INTO %s SELECT a1, c1, d1, e1, a2, e2 FROM %s INNER JOIN %s ON c1 = c2 AND d1 = d2 AND e1 = e2",
+ sinkTableName, leftTableName, rightTableName);
+
+ assertThat(tEnv.explainSql(sql))
+ .contains(
+ "DeltaJoin(joinType=[InnerJoin], where=[((c1 = c2) AND (d1 = d2) AND (e1 = e2))]");
+
+ tEnv.executeSql(sql);
+
+ CloseableIterator<Row> collected =
+ tEnv.executeSql(String.format("select * from %s", sinkTableName)).collect();
+ // Only first row should match (e1 = e2 = 10000)
+ // Second row filtered out (e1 = 20000 AND != e2 = 99999)
+ List<String> expected =
+ Arrays.asList(
+ "+I[1, 100, 1, 10000, 1, 10000]",
+ "-U[1, 100, 1, 10000, 1, 10000]",
+ "+U[1, 100, 1, 10000, 1, 10000]");
+ assertResultsIgnoreOrder(collected, expected, true);
+ }
+
+ @Test
+ void testDeltaJoinFailsWhenJoinKeyNotContainIndex() {
+ disableSinkRequireOnConflict();
+ String leftTableName = "left_table_no_idx_force";
+ createSource(
+ leftTableName,
+ "a1 int, b1 varchar, c1 bigint, d1 int, e1 bigint",
+ "c1, d1",
+ "c1",
+ null,
+ ImmutableMap.of("table.delete.behavior", "IGNORE"));
+
+ String rightTableName = "right_table_no_idx_force";
+ createSource(
+ rightTableName,
+ "a2 int, b2 varchar, c2 bigint, d2 int, e2 bigint",
+ "c2, d2",
+ "c2",
+ null,
+ ImmutableMap.of("table.delete.behavior", "IGNORE"));
+
+ String sinkTableName = "sink_table_no_idx_force";
+ createSink(
+ sinkTableName,
+ "a1 int, b1 varchar, c1 bigint, d1 int, e1 bigint, a2 int, b2 varchar, c2 bigint, d2 int, e2 bigint",
+ "a1, a2");
+
+ String sql =
+ String.format(
+ "INSERT INTO %s SELECT * FROM %s INNER JOIN %s ON a1 = a2",
+ sinkTableName, leftTableName, rightTableName);
+
+ assertThatThrownBy(() -> tEnv.explainSql(sql))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("doesn't support to do delta join optimization");
+ }
+
+ @Test
+ void testDeltaJoinFailsWithOuterJoin() {
+ disableSinkRequireOnConflict();
+ String leftTableName = "left_table_outer_fail";
+ createSource(
+ leftTableName,
+ "a1 int, c1 bigint, d1 int",
+ "c1, d1",
+ "c1",
+ null,
+ ImmutableMap.of("table.delete.behavior", "IGNORE"));
+
+ String rightTableName = "right_table_outer_fail";
+ createSource(
+ rightTableName,
+ "a2 int, c2 bigint, d2 int",
+ "c2, d2",
+ "c2",
+ null,
+ ImmutableMap.of("table.delete.behavior", "IGNORE"));
+
+ String sinkTableName = "sink_table_outer_fail";
+ createSink(sinkTableName, "a1 int, c1 bigint, a2 int", "c1");
+
+ // Test LEFT JOIN
+ String leftJoinSql =
+ String.format(
+ "INSERT INTO %s SELECT a1, c1, a2 FROM %s LEFT JOIN %s ON c1 = c2 AND d1 = d2",
+ sinkTableName, leftTableName, rightTableName);
+
+ assertThatThrownBy(() -> tEnv.explainSql(leftJoinSql))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("doesn't support to do delta join optimization");
+
+ // Test RIGHT JOIN
+ String rightJoinSql =
+ String.format(
+ "INSERT INTO %s SELECT a1, c2, a2 FROM %s RIGHT JOIN %s ON c1 = c2 AND d1 = d2",
+ sinkTableName, leftTableName, rightTableName);
+
+ assertThatThrownBy(() -> tEnv.explainSql(rightJoinSql))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("doesn't support to do delta join optimization");
+
+ // Test FULL OUTER JOIN
+ String fullOuterJoinSql =
+ String.format(
+ "INSERT INTO %s SELECT a1, c1, a2 FROM %s FULL OUTER JOIN %s ON c1 = c2 AND d1 = d2",
+ sinkTableName, leftTableName, rightTableName);
+
+ assertThatThrownBy(() -> tEnv.explainSql(fullOuterJoinSql))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("doesn't support to do delta join optimization");
+ }
+
+ @Test
+ void testDeltaJoinFailsWithSinkMaterializer() {
+ disableSinkRequireOnConflict();
+ // With CDC sources, when sink PK doesn't match upstream update key,
+ // Flink would insert SinkUpsertMaterializer which prevents delta join
+ String leftTableName = "left_table_materializer";
+ createSource(
+ leftTableName,
+ "a1 int, b1 varchar, c1 bigint, d1 int, e1 bigint",
+ "c1, d1",
+ "c1",
+ null,
+ ImmutableMap.of("table.delete.behavior", "IGNORE"));
+
+ String rightTableName = "right_table_materializer";
+ createSource(
+ rightTableName,
+ "a2 int, b2 varchar, c2 bigint, d2 int, e2 bigint",
+ "c2, d2",
+ "c2",
+ null,
+ ImmutableMap.of("table.delete.behavior", "IGNORE"));
+
+ // Sink PK (a1, a2) doesn't match upstream update key (c1, d1, c2, d2)
+ // TODO: this depends on Fluss supports MVCC/point-in-time lookup to support change upsert
+ // keys
+ String sinkTableName = "sink_table_materializer";
+ createSink(
+ sinkTableName,
+ "a1 int, b1 varchar, c1 bigint, d1 int, e1 bigint, a2 int, b2 varchar, c2 bigint, d2 int, e2 bigint",
+ "a1, a2");
+
+ String sql =
+ String.format(
+ "INSERT INTO %s SELECT * FROM %s INNER JOIN %s ON c1 = c2 AND d1 = d2",
+ sinkTableName, leftTableName, rightTableName);
+
+ assertThatThrownBy(() -> tEnv.explainSql(sql))
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("doesn't support to do delta join optimization");
+ }
+
+ /**
+ * Documents the SQL shape for a "lookup join after a delta join" pipeline (Flink 2.3): a delta
+ * join wrapped in a subquery that adds a {@code PROCTIME()} time attribute, followed by a
+ * downstream lookup (temporal) join against a Fluss dimension table. The proctime computed via
+ * {@code PROCTIME()} in the inner subquery is consumed by the {@code FOR SYSTEM_TIME AS OF}
+ * clause of the downstream lookup join.
+ *
+ * <p>Per Flink 2.3 "Supported Features and Limitations" docs: "Support for lookup join after a
+ * delta join" and "Support for non-deterministic functions in projections and filters between
+ * the delta join and downstream operators".
+ *
+ * <p>Today this is asserted as an expected failure to lock in the precise planner error that is
+ * raised today: the {@code StreamPhysicalSnapshotOnCalcTableScanRule} in Flink 2.3 invokes
+ * {@code FlussTableSource#getLookupRuntimeProvider} with an empty {@code
+ * LookupContext.getKeys()} for the dim table in this scenario, and {@code
+ * LookupNormalizer.validateAndCreateLookupNormalizer} then throws "Can't find expected key 'id'
+ * in lookup keys []". The delta-join strategy is overridden to {@link
+ * OptimizerConfigOptions.DeltaJoinStrategy#AUTO} so that the test is not blocked by {@code
+ * StreamPhysicalDeltaJoinForceValidator}.
+ */
+ @Test
+ void testLookupJoinAfterDeltaJoin() {
+ // allow the planner to choose; the validator only rejects when FORCE is set
+ tEnv.getConfig()
+ .set(
+ OptimizerConfigOptions.TABLE_OPTIMIZER_DELTA_JOIN_STRATEGY,
+ OptimizerConfigOptions.DeltaJoinStrategy.AUTO);
+ String leftTableName = "left_table_lj_delta";
+ String rightTableName = "right_table_lj_delta";
+ String dimTableName = "dim_table_lj_delta";
+ String sinkTableName = "sink_table_lj_delta";
+
+ createSource(
+ leftTableName,
+ "a1 int, b1 varchar, c1 bigint, d1 int, e1 bigint",
+ "c1, d1",
+ "c1",
+ null,
+ ImmutableMap.of("table.delete.behavior", "IGNORE"));
+ createSource(
+ rightTableName,
+ "a2 int, b2 varchar, c2 bigint, d2 int, e2 bigint",
+ "c2, d2",
+ "c2",
+ null,
+ ImmutableMap.of("table.delete.behavior", "IGNORE"));
+
+ tEnv.executeSql(
+ String.format(
+ "create table %s ("
+ + " id int not null,"
+ + " name varchar,"
+ + " primary key (id) NOT ENFORCED"
+ + ") with ("
+ + " 'connector' = 'fluss',"
+ + " 'bucket.num' = '1'"
+ + ")",
+ dimTableName));
+
+ createSink(
+ sinkTableName,
+ "a1 int, b1 varchar, c1 bigint, d1 int, a2 int, dim_name varchar",
+ "c1, d1");
+
+ String sql =
+ String.format(
+ "INSERT INTO %s "
+ + "SELECT tmp.a1, tmp.b1, tmp.c1, tmp.d1, tmp.a2, dim.name "
+ + "FROM ("
+ + " SELECT l.*, r.a2, PROCTIME() AS proc "
+ + " FROM %s AS l INNER JOIN %s AS r ON l.c1 = r.c2"
+ + ") tmp "
+ + "JOIN %s FOR SYSTEM_TIME AS OF tmp.proc AS dim "
+ + " ON dim.id = tmp.c1 "
+ + "ON CONFLICT DO DEDUPLICATE",
+ sinkTableName, leftTableName, rightTableName, dimTableName);
+
+ assertThatThrownBy(() -> tEnv.explainSql(sql))
+ .hasRootCauseInstanceOf(org.apache.flink.table.api.TableException.class)
+ .hasRootCauseMessage(
+ "The Fluss lookup function supports lookup tables where the lookup keys"
+ + " include all primary keys or all bucket keys. Can't find"
+ + " expected key 'id' in lookup keys []");
+ }
+}
diff --git a/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/source/Flink23TableSourceBatchITCase.java b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/source/Flink23TableSourceBatchITCase.java
new file mode 100644
index 000000000..0df267e85
--- /dev/null
+++ b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/source/Flink23TableSourceBatchITCase.java
@@ -0,0 +1,21 @@
+/*
+ * 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.fluss.flink.source;
+
+/** IT case for batch source in Flink 2.3. */
+public class Flink23TableSourceBatchITCase extends FlinkTableSourceBatchITCase {}
diff --git a/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/source/Flink23TableSourceFailOverITCase.java b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/source/Flink23TableSourceFailOverITCase.java
new file mode 100644
index 000000000..37e7b53f6
--- /dev/null
+++ b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/source/Flink23TableSourceFailOverITCase.java
@@ -0,0 +1,21 @@
+/*
+ * 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.fluss.flink.source;
+
+/** IT case for source failover and recovery in Flink 2.3. */
+public class Flink23TableSourceFailOverITCase extends FlinkTableSourceFailOverITCase {}
diff --git a/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/source/Flink23TableSourceITCase.java b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/source/Flink23TableSourceITCase.java
new file mode 100644
index 000000000..bbe34f153
--- /dev/null
+++ b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/source/Flink23TableSourceITCase.java
@@ -0,0 +1,21 @@
+/*
+ * 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.fluss.flink.source;
+
+/** IT case for {@link FlinkTableSource} in Flink 2.3. */
+public class Flink23TableSourceITCase extends FlinkTableSourceITCase {}
diff --git a/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/tiering/Flink23TieringITCase.java b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/tiering/Flink23TieringITCase.java
new file mode 100644
index 000000000..91db3a6f2
--- /dev/null
+++ b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/tiering/Flink23TieringITCase.java
@@ -0,0 +1,22 @@
+/*
+ * 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.fluss.flink.tiering;
+
+/** The IT case for tiering in Flink 2.3. */
+class Flink23TieringITCase extends TieringITCase {}
diff --git a/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/tiering/committer/Flink23TieringCommitOperatorTest.java b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/tiering/committer/Flink23TieringCommitOperatorTest.java
new file mode 100644
index 000000000..2f6d5d677
--- /dev/null
+++ b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/tiering/committer/Flink23TieringCommitOperatorTest.java
@@ -0,0 +1,24 @@
+/*
+ * 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.fluss.flink.tiering.committer;
+
+/**
+ * UT for {@link TieringCommitOperator}. Test the compatibility of the `getAttemptNumber` method in
+ * flink 2.3.
+ */
+public class Flink23TieringCommitOperatorTest extends TieringCommitOperatorTest {}
diff --git a/fluss-flink/fluss-flink-2.3/src/test/resources/META-INF/services/org.junit.jupiter.api.extension.Extension b/fluss-flink/fluss-flink-2.3/src/test/resources/META-INF/services/org.junit.jupiter.api.extension.Extension
new file mode 100644
index 000000000..ca0e907f6
--- /dev/null
+++ b/fluss-flink/fluss-flink-2.3/src/test/resources/META-INF/services/org.junit.jupiter.api.extension.Extension
@@ -0,0 +1,19 @@
+#
+# 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.fluss.testutils.common.TestLoggerExtension
\ No newline at end of file
diff --git a/fluss-flink/fluss-flink-2.3/src/test/resources/junit-platform.properties b/fluss-flink/fluss-flink-2.3/src/test/resources/junit-platform.properties
new file mode 100644
index 000000000..d5b31f139
--- /dev/null
+++ b/fluss-flink/fluss-flink-2.3/src/test/resources/junit-platform.properties
@@ -0,0 +1,20 @@
+#
+# 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.
+#
+
+junit.jupiter.extensions.autodetection.enabled=true
+fluss.tests.multiversion.enabled=true
diff --git a/fluss-flink/fluss-flink-2.3/src/test/resources/log4j2-test.properties b/fluss-flink/fluss-flink-2.3/src/test/resources/log4j2-test.properties
new file mode 100644
index 000000000..12b05f186
--- /dev/null
+++ b/fluss-flink/fluss-flink-2.3/src/test/resources/log4j2-test.properties
@@ -0,0 +1,32 @@
+#
+# 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
+
+# suppress the duplicated logger extension
+logger.flink.name = org.apache.flink.util.TestLoggerExtension
+logger.flink.level = OFF
diff --git a/fluss-flink/fluss-flink-common/README.md b/fluss-flink/fluss-flink-common/README.md
index 33ca64f30..e1455cee6 100644
--- a/fluss-flink/fluss-flink-common/README.md
+++ b/fluss-flink/fluss-flink-common/README.md
@@ -28,12 +28,13 @@ classes and regular unit test classes.
## Execution Model
-| Flink module | Tests that run |
-| --- | --- |
+| Flink module | Tests that run |
+|--------------------| --- |
| `fluss-flink-1.20` | All tests |
| `fluss-flink-1.18` | Only tests marked with `@MultiVersionTest` |
| `fluss-flink-1.19` | Only tests marked with `@MultiVersionTest` |
-| `fluss-flink-2.2` | Only tests marked with `@MultiVersionTest` |
+| `fluss-flink-2.2` | Only tests marked with `@MultiVersionTest` |
+| `fluss-flink-2.3` | Only tests marked with `@MultiVersionTest` |
The non-default modules enable JUnit extension auto-detection and multi-version filtering in their
`src/test/resources/junit-platform.properties` files:
@@ -178,7 +179,7 @@ Run all filtered compatibility suites:
```bash
./mvnw verify \
- -pl fluss-flink/fluss-flink-1.18,fluss-flink/fluss-flink-1.19,fluss-flink/fluss-flink-2.2 \
+ -pl fluss-flink/fluss-flink-1.18,fluss-flink/fluss-flink-1.19,fluss-flink/fluss-flink-2.2,fluss-flink/fluss-flink-2.3 \
-am
```
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/FlinkConnectorOptions.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/FlinkConnectorOptions.java
index 23d8f0b2e..eddc20629 100644
--- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/FlinkConnectorOptions.java
+++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/FlinkConnectorOptions.java
@@ -26,7 +26,6 @@ import org.apache.flink.configuration.ConfigOptions;
import org.apache.flink.configuration.DescribedEnum;
import org.apache.flink.configuration.description.InlineElement;
import org.apache.flink.table.catalog.CatalogMaterializedTable;
-import org.apache.flink.table.catalog.IntervalFreshness;
import java.time.Duration;
import java.util.Arrays;
@@ -249,18 +248,29 @@ public class FlinkConnectorOptions {
.noDefaultValue()
.withDescription(
"The definition query text of materialized table, text is expanded in contrast to the original SQL.");
+ public static final ConfigOption<String> MATERIALIZED_TABLE_ORIGINAL_QUERY =
+ ConfigOptions.key("materialized-table.original-query")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Original text of the materialized table definition that preserves the original formatting.");
+ public static final ConfigOption<String> MATERIALIZED_TABLE_EXPANDED_QUERY =
+ ConfigOptions.key("materialized-table.expanded-query")
+ .stringType()
+ .noDefaultValue()
+ .withDescription(
+ "Expanded text of the original materialized table definition with resolved identifiers.");
public static final ConfigOption<String> MATERIALIZED_TABLE_INTERVAL_FRESHNESS =
ConfigOptions.key("materialized-table.interval-freshness")
.stringType()
.noDefaultValue()
.withDescription(
"The freshness interval of materialized table which is used to determine the physical refresh mode.");
- public static final ConfigOption<IntervalFreshness.TimeUnit>
- MATERIALIZED_TABLE_INTERVAL_FRESHNESS_TIME_UNIT =
- ConfigOptions.key("materialized-table.interval-freshness.time-unit")
- .enumType(IntervalFreshness.TimeUnit.class)
- .noDefaultValue()
- .withDescription("The time unit of freshness interval.");
+ public static final ConfigOption<String> MATERIALIZED_TABLE_INTERVAL_FRESHNESS_TIME_UNIT =
+ ConfigOptions.key("materialized-table.interval-freshness.time-unit")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("The time unit of freshness interval.");
public static final ConfigOption<CatalogMaterializedTable.LogicalRefreshMode>
MATERIALIZED_TABLE_LOGICAL_REFRESH_MODE =
ConfigOptions.key("materialized-table.logical-refresh-mode")
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/adapter/CatalogMaterializedTableAdapter.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/adapter/CatalogMaterializedTableAdapter.java
new file mode 100644
index 000000000..f7e0df853
--- /dev/null
+++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/adapter/CatalogMaterializedTableAdapter.java
@@ -0,0 +1,131 @@
+/*
+ * 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.fluss.flink.adapter;
+
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.catalog.CatalogMaterializedTable;
+import org.apache.flink.table.catalog.IntervalFreshness;
+import org.apache.flink.table.catalog.TableDistribution;
+
+import javax.annotation.Nullable;
+
+import java.util.List;
+import java.util.Map;
+
+/** An adapter for {@link CatalogMaterializedTable#newBuilder()} constructor for flink1.x. */
+public class CatalogMaterializedTableAdapter {
+
+ private final CatalogMaterializedTable.Builder builder;
+
+ private CatalogMaterializedTableAdapter() {
+ this.builder = CatalogMaterializedTable.newBuilder();
+ }
+
+ public static CatalogMaterializedTableAdapter newAdapter() {
+ return new CatalogMaterializedTableAdapter();
+ }
+
+ public CatalogMaterializedTableAdapter schema(Schema schema) {
+ this.builder.schema(schema);
+ return this;
+ }
+
+ public CatalogMaterializedTableAdapter comment(@Nullable String comment) {
+ this.builder.comment(comment);
+ return this;
+ }
+
+ public CatalogMaterializedTableAdapter partitionKeys(List<String> partitionKeys) {
+ this.builder.partitionKeys(partitionKeys);
+ return this;
+ }
+
+ public CatalogMaterializedTableAdapter options(Map<String, String> options) {
+ this.builder.options(options);
+ return this;
+ }
+
+ public CatalogMaterializedTableAdapter snapshot(@Nullable Long snapshot) {
+ this.builder.snapshot(snapshot);
+ return this;
+ }
+
+ public CatalogMaterializedTableAdapter originalQuery(String originalQuery) {
+ return this;
+ }
+
+ public CatalogMaterializedTableAdapter expandedQuery(String expandedQuery) {
+ return this;
+ }
+
+ public CatalogMaterializedTableAdapter definitionQuery(String definitionQuery) {
+ this.builder.definitionQuery(definitionQuery);
+ return this;
+ }
+
+ public CatalogMaterializedTableAdapter freshness(@Nullable IntervalFreshness freshness) {
+ this.builder.freshness(freshness);
+ return this;
+ }
+
+ public CatalogMaterializedTableAdapter logicalRefreshMode(
+ CatalogMaterializedTable.LogicalRefreshMode logicalRefreshMode) {
+ this.builder.logicalRefreshMode(logicalRefreshMode);
+ return this;
+ }
+
+ public CatalogMaterializedTableAdapter refreshMode(
+ @Nullable CatalogMaterializedTable.RefreshMode refreshMode) {
+ this.builder.refreshMode(refreshMode);
+ return this;
+ }
+
+ public CatalogMaterializedTableAdapter refreshStatus(
+ CatalogMaterializedTable.RefreshStatus refreshStatus) {
+ this.builder.refreshStatus(refreshStatus);
+ return this;
+ }
+
+ public CatalogMaterializedTableAdapter refreshHandlerDescription(
+ @Nullable String refreshHandlerDescription) {
+ this.builder.refreshHandlerDescription(refreshHandlerDescription);
+ return this;
+ }
+
+ public CatalogMaterializedTableAdapter serializedRefreshHandler(
+ @Nullable byte[] serializedRefreshHandler) {
+ this.builder.serializedRefreshHandler(serializedRefreshHandler);
+ return this;
+ }
+
+ public CatalogMaterializedTableAdapter distribution(@Nullable TableDistribution distribution) {
+ return this;
+ }
+
+ public CatalogMaterializedTable build() {
+ return this.builder.build();
+ }
+
+ public static String getExpandedQuery(CatalogMaterializedTable mt) {
+ return null;
+ }
+
+ public static String getOriginalQuery(CatalogMaterializedTable mt) {
+ return null;
+ }
+}
diff --git a/fluss-flink/fluss-flink-2.2/src/test/java/org/apache/fluss/flink/adapter/ResolvedCatalogMaterializedTableAdapter.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/adapter/IntervalFreshnessAdapter.java
similarity index 50%
rename from fluss-flink/fluss-flink-2.2/src/test/java/org/apache/fluss/flink/adapter/ResolvedCatalogMaterializedTableAdapter.java
rename to fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/adapter/IntervalFreshnessAdapter.java
index f868a84db..4ae00a1b1 100644
--- a/fluss-flink/fluss-flink-2.2/src/test/java/org/apache/fluss/flink/adapter/ResolvedCatalogMaterializedTableAdapter.java
+++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/adapter/IntervalFreshnessAdapter.java
@@ -17,25 +17,29 @@
package org.apache.fluss.flink.adapter;
-import org.apache.flink.table.catalog.CatalogMaterializedTable;
import org.apache.flink.table.catalog.IntervalFreshness;
-import org.apache.flink.table.catalog.ResolvedCatalogMaterializedTable;
-import org.apache.flink.table.catalog.ResolvedSchema;
-/**
- * Adapter for {@link ResolvedCatalogMaterializedTable} because the constructor is compatibility in
- * flink 2.2. However, this constructor only used in test.
- *
- * <p>TODO: remove it until <a href="https://issues.apache.org/jira/browse/FLINK-38532">...</a> is
- * fixed.
- */
-public class ResolvedCatalogMaterializedTableAdapter {
-
- public static ResolvedCatalogMaterializedTable create(
- CatalogMaterializedTable origin,
- ResolvedSchema resolvedSchema,
- CatalogMaterializedTable.RefreshMode refreshMode,
- IntervalFreshness freshness) {
- return new ResolvedCatalogMaterializedTable(origin, resolvedSchema, refreshMode, freshness);
+/** An adapter for {@link IntervalFreshness} for Flink versions below 2.3. */
+public class IntervalFreshnessAdapter {
+
+ public static TimeUnitAdapter timeUnit(String name) {
+ return new TimeUnitAdapter(IntervalFreshness.TimeUnit.valueOf(name));
+ }
+
+ public static IntervalFreshness of(String interval, TimeUnitAdapter timeUnit) {
+ return IntervalFreshness.of(interval, timeUnit.timeUnit);
+ }
+
+ public static String getTimeUnitName(IntervalFreshness intervalFreshness) {
+ return intervalFreshness.getTimeUnit().name();
+ }
+
+ /** An adapter for {@link IntervalFreshness.TimeUnit} for Flink versions below 2.3. */
+ public static class TimeUnitAdapter {
+ private final IntervalFreshness.TimeUnit timeUnit;
+
+ private TimeUnitAdapter(IntervalFreshness.TimeUnit timeUnit) {
+ this.timeUnit = timeUnit;
+ }
}
}
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/FlinkConversions.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/FlinkConversions.java
index 5c2592292..c200db857 100644
--- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/FlinkConversions.java
+++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/FlinkConversions.java
@@ -21,7 +21,9 @@ import org.apache.fluss.annotation.VisibleForTesting;
import org.apache.fluss.config.ConfigOption;
import org.apache.fluss.config.MemorySize;
import org.apache.fluss.config.Password;
+import org.apache.fluss.flink.adapter.CatalogMaterializedTableAdapter;
import org.apache.fluss.flink.adapter.CatalogTableAdapter;
+import org.apache.fluss.flink.adapter.IntervalFreshnessAdapter;
import org.apache.fluss.flink.catalog.FlinkCatalogFactory;
import org.apache.fluss.metadata.AggFunction;
import org.apache.fluss.metadata.DatabaseDescriptor;
@@ -77,9 +79,11 @@ import static org.apache.fluss.flink.FlinkConnectorOptions.AUTO_INCREMENT_FIELDS
import static org.apache.fluss.flink.FlinkConnectorOptions.BUCKET_KEY;
import static org.apache.fluss.flink.FlinkConnectorOptions.BUCKET_NUMBER;
import static org.apache.fluss.flink.FlinkConnectorOptions.MATERIALIZED_TABLE_DEFINITION_QUERY;
+import static org.apache.fluss.flink.FlinkConnectorOptions.MATERIALIZED_TABLE_EXPANDED_QUERY;
import static org.apache.fluss.flink.FlinkConnectorOptions.MATERIALIZED_TABLE_INTERVAL_FRESHNESS;
import static org.apache.fluss.flink.FlinkConnectorOptions.MATERIALIZED_TABLE_INTERVAL_FRESHNESS_TIME_UNIT;
import static org.apache.fluss.flink.FlinkConnectorOptions.MATERIALIZED_TABLE_LOGICAL_REFRESH_MODE;
+import static org.apache.fluss.flink.FlinkConnectorOptions.MATERIALIZED_TABLE_ORIGINAL_QUERY;
import static org.apache.fluss.flink.FlinkConnectorOptions.MATERIALIZED_TABLE_PREFIX;
import static org.apache.fluss.flink.FlinkConnectorOptions.MATERIALIZED_TABLE_REFRESH_HANDLER_BYTES;
import static org.apache.fluss.flink.FlinkConnectorOptions.MATERIALIZED_TABLE_REFRESH_HANDLER_DESCRIPTION;
@@ -597,12 +601,23 @@ public class FlinkConversions {
CatalogMaterializedTable mt, Map<String, String> customProperties) {
// Serialize core materialized table properties
customProperties.put(MATERIALIZED_TABLE_DEFINITION_QUERY.key(), mt.getDefinitionQuery());
+ // Only persist original/expanded queries when the active Flink version actually exposes
+ // them (Flink 2.3+). Older versions' adapter returns null; persisting a null would
+ // break downstream readers that route customProperties through Configuration.setString.
+ String originalQuery = CatalogMaterializedTableAdapter.getOriginalQuery(mt);
+ if (originalQuery != null) {
+ customProperties.put(MATERIALIZED_TABLE_ORIGINAL_QUERY.key(), originalQuery);
+ }
+ String expandedQuery = CatalogMaterializedTableAdapter.getExpandedQuery(mt);
+ if (expandedQuery != null) {
+ customProperties.put(MATERIALIZED_TABLE_EXPANDED_QUERY.key(), expandedQuery);
+ }
// Serialize freshness configuration
IntervalFreshness freshness = mt.getDefinitionFreshness();
customProperties.put(MATERIALIZED_TABLE_INTERVAL_FRESHNESS.key(), freshness.getInterval());
customProperties.put(
MATERIALIZED_TABLE_INTERVAL_FRESHNESS_TIME_UNIT.key(),
- freshness.getTimeUnit().name());
+ IntervalFreshnessAdapter.getTimeUnitName(freshness));
// Serialize refresh configuration
customProperties.put(
MATERIALIZED_TABLE_LOGICAL_REFRESH_MODE.key(), mt.getLogicalRefreshMode().name());
@@ -643,6 +658,8 @@ public class FlinkConversions {
Map<String, String> options) {
// Validate required materialized table options first
String definitionQuery = options.get(MATERIALIZED_TABLE_DEFINITION_QUERY.key());
+ String originalQuery = options.get(MATERIALIZED_TABLE_ORIGINAL_QUERY.key());
+ String expandedQuery = options.get(MATERIALIZED_TABLE_EXPANDED_QUERY.key());
String intervalFreshness = options.get(MATERIALIZED_TABLE_INTERVAL_FRESHNESS.key());
String timeUnitStr = options.get(MATERIALIZED_TABLE_INTERVAL_FRESHNESS_TIME_UNIT.key());
String logicalRefreshModeStr = options.get(MATERIALIZED_TABLE_LOGICAL_REFRESH_MODE.key());
@@ -663,9 +680,21 @@ public class FlinkConversions {
checkNotNull(refreshModeStr, "Materialized table refresh mode is required but missing");
checkNotNull(refreshStatusStr, "Materialized table refresh status is required but missing");
+ // Compatibility: materialized tables persisted before Flink 2.3 only stored the
+ // definition-query. Flink 2.3's DefaultCatalogMaterializedTable rejects null
+ // original/expanded queries, so fall back to definition-query when those keys are
+ // absent — the closest approximation we can recover from a legacy payload.
+ if (originalQuery == null) {
+ originalQuery = definitionQuery;
+ }
+ if (expandedQuery == null) {
+ expandedQuery = definitionQuery;
+ }
+
// Parse validated values
- IntervalFreshness.TimeUnit timeUnit = IntervalFreshness.TimeUnit.valueOf(timeUnitStr);
- IntervalFreshness freshness = IntervalFreshness.of(intervalFreshness, timeUnit);
+ IntervalFreshnessAdapter.TimeUnitAdapter timeUnit =
+ IntervalFreshnessAdapter.timeUnit(timeUnitStr);
+ IntervalFreshness freshness = IntervalFreshnessAdapter.of(intervalFreshness, timeUnit);
CatalogMaterializedTable.LogicalRefreshMode logicalRefreshMode =
CatalogMaterializedTable.LogicalRefreshMode.valueOf(logicalRefreshModeStr);
CatalogMaterializedTable.RefreshMode refreshMode =
@@ -685,12 +714,14 @@ public class FlinkConversions {
? null
: decodeBase64ToBytes(refreshHandlerStringBytes);
- CatalogMaterializedTable.Builder builder = CatalogMaterializedTable.newBuilder();
+ CatalogMaterializedTableAdapter builder = CatalogMaterializedTableAdapter.newAdapter();
builder.schema(schema)
.comment(comment)
.partitionKeys(partitionKeys)
.options(excludeByPrefix(options, MATERIALIZED_TABLE_PREFIX))
.definitionQuery(definitionQuery)
+ .originalQuery(originalQuery)
+ .expandedQuery(expandedQuery)
.freshness(freshness)
.logicalRefreshMode(logicalRefreshMode)
.refreshMode(refreshMode)
diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogTest.java
index f53fcec21..88f520bbc 100644
--- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogTest.java
+++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogTest.java
@@ -22,7 +22,7 @@ import org.apache.fluss.config.Configuration;
import org.apache.fluss.exception.IllegalConfigurationException;
import org.apache.fluss.exception.InvalidPartitionException;
import org.apache.fluss.exception.InvalidTableException;
-import org.apache.fluss.flink.adapter.ResolvedCatalogMaterializedTableAdapter;
+import org.apache.fluss.flink.adapter.CatalogMaterializedTableAdapter;
import org.apache.fluss.flink.lake.LakeFlinkCatalog;
import org.apache.fluss.flink.utils.FlinkConversionsTest;
import org.apache.fluss.server.testutils.FlussClusterExtension;
@@ -42,6 +42,7 @@ import org.apache.flink.table.catalog.Column;
import org.apache.flink.table.catalog.GenericInMemoryCatalog;
import org.apache.flink.table.catalog.IntervalFreshness;
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.TableChange;
@@ -145,13 +146,16 @@ class FlinkCatalogTest {
CatalogMaterializedTable.RefreshMode refreshMode,
Map<String, String> options) {
CatalogMaterializedTable origin =
- CatalogMaterializedTable.newBuilder()
+ CatalogMaterializedTableAdapter.newAdapter()
.schema(Schema.newBuilder().fromResolvedSchema(resolvedSchema).build())
.comment("test comment")
.options(options)
.partitionKeys(Collections.emptyList())
.definitionQuery("select first, second, third from t")
- .freshness(IntervalFreshness.of("5", IntervalFreshness.TimeUnit.SECOND))
+ // TODO The configuration added in Flink 2.3 currently uses the same SQL.
+ .originalQuery("select first, second, third from t")
+ .expandedQuery("select first, second, third from t")
+ .freshness(IntervalFreshness.ofSecond("5"))
.logicalRefreshMode(
refreshMode == CatalogMaterializedTable.RefreshMode.CONTINUOUS
? CatalogMaterializedTable.LogicalRefreshMode.CONTINUOUS
@@ -159,11 +163,16 @@ class FlinkCatalogTest {
.refreshMode(refreshMode)
.refreshStatus(CatalogMaterializedTable.RefreshStatus.INITIALIZING)
.build();
- return ResolvedCatalogMaterializedTableAdapter.create(
- origin,
- resolvedSchema,
- refreshMode,
- IntervalFreshness.of("5", IntervalFreshness.TimeUnit.SECOND));
+ return createResolvedCatalogMaterializedTable(
+ origin, resolvedSchema, refreshMode, IntervalFreshness.ofSecond("5"));
+ }
+
+ protected ResolvedCatalogMaterializedTable createResolvedCatalogMaterializedTable(
+ CatalogMaterializedTable origin,
+ ResolvedSchema resolvedSchema,
+ CatalogMaterializedTable.RefreshMode refreshMode,
+ IntervalFreshness intervalFreshness) {
+ return new ResolvedCatalogMaterializedTable(origin, resolvedSchema);
}
protected FlinkCatalog initCatalog(
diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/FlinkTableSinkITCase.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/FlinkTableSinkITCase.java
index 2015bece2..09a59de57 100644
--- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/FlinkTableSinkITCase.java
+++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/sink/FlinkTableSinkITCase.java
@@ -532,6 +532,7 @@ abstract class FlinkTableSinkITCase extends AbstractTestBase {
@Test
@MultiVersionTest
void testPartialUpsert() throws Exception {
+ disableSinkRequireOnConflict();
tEnv.executeSql(
"create table sink_test (a int not null primary key not enforced, b bigint, c string) with('bucket.num' = '3')");
@@ -697,7 +698,11 @@ abstract class FlinkTableSinkITCase extends AbstractTestBase {
@ParameterizedTest
@ValueSource(booleans = {true, false})
+ @MultiVersionTest
void testIgnoreDelete(boolean isPrimaryKeyTable) throws Exception {
+ if (isPrimaryKeyTable) {
+ disableSinkRequireOnConflict();
+ }
String sinkName =
isPrimaryKeyTable
? "ignore_delete_primary_key_table_sink"
@@ -1987,4 +1992,14 @@ abstract class FlinkTableSinkITCase extends AbstractTestBase {
tableName),
expectedResults);
}
+
+ /**
+ * Hook for tests that need the Flink option {@code table.exec.sink.require-on-conflict} (Flink
+ * 2.3+; lives on {@code org.apache.flink.table.api.config.ExecutionConfigOptions} in that
+ * version) disabled. Flink 2.3 enables it by default.
+ *
+ * <p>This base implementation is a no-op so the same test code runs unchanged on every Flink
+ * version. The Flink 2.3-specific subclass overrides it to actually flip the option off.
+ */
+ protected void disableSinkRequireOnConflict() {}
}
diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/FlinkConversionsTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/FlinkConversionsTest.java
index a9d542a28..71a81d3a0 100644
--- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/FlinkConversionsTest.java
+++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/FlinkConversionsTest.java
@@ -55,6 +55,14 @@ import static org.apache.flink.table.api.DataTypes.VARBINARY;
import static org.apache.flink.table.api.DataTypes.VARCHAR;
import static org.apache.fluss.flink.FlinkConnectorOptions.BUCKET_KEY;
import static org.apache.fluss.flink.FlinkConnectorOptions.BUCKET_NUMBER;
+import static org.apache.fluss.flink.FlinkConnectorOptions.MATERIALIZED_TABLE_DEFINITION_QUERY;
+import static org.apache.fluss.flink.FlinkConnectorOptions.MATERIALIZED_TABLE_EXPANDED_QUERY;
+import static org.apache.fluss.flink.FlinkConnectorOptions.MATERIALIZED_TABLE_INTERVAL_FRESHNESS;
+import static org.apache.fluss.flink.FlinkConnectorOptions.MATERIALIZED_TABLE_INTERVAL_FRESHNESS_TIME_UNIT;
+import static org.apache.fluss.flink.FlinkConnectorOptions.MATERIALIZED_TABLE_LOGICAL_REFRESH_MODE;
+import static org.apache.fluss.flink.FlinkConnectorOptions.MATERIALIZED_TABLE_ORIGINAL_QUERY;
+import static org.apache.fluss.flink.FlinkConnectorOptions.MATERIALIZED_TABLE_REFRESH_MODE;
+import static org.apache.fluss.flink.FlinkConnectorOptions.MATERIALIZED_TABLE_REFRESH_STATUS;
import static org.apache.fluss.flink.utils.CatalogTableTestUtils.addOptions;
import static org.apache.fluss.flink.utils.CatalogTableTestUtils.checkEqualsIgnoreSchema;
import static org.apache.fluss.record.TestData.DEFAULT_REMOTE_DATA_DIR;
@@ -529,6 +537,137 @@ public class FlinkConversionsTest {
assertThat(convertedFlinkTable.getOptions()).containsAllEntriesOf(options);
}
+ /**
+ * Verifies that {@link FlinkConversions#toFlinkTable} can read a materialized-table payload
+ * that only carries {@code definition-query} — the shape persisted by Fluss before Flink 2.3
+ * introduced {@code original-query}/{@code expanded-query}. The deserialization must not fail,
+ * the definition-query must survive, and the three query keys must be consumed from the
+ * resulting options map.
+ *
+ * <p>The {@code fluss-flink-common} test classpath uses the Flink 1.20/2.2 adapter, where the
+ * {@code originalQuery}/{@code expandedQuery} setters are no-ops, so this test only asserts the
+ * public surface: definition-query is preserved, and the three query keys are not leaked into
+ * {@code getOptions()}. Verifying that the Flink 2.3 adapter actually populates the new fields
+ * with the fallback value is covered by the dedicated test in {@code Flink23CatalogTest}.
+ */
+ @Test
+ void testMaterializedTableFallsBackToDefinitionQueryForLegacyData() {
+ String definitionQuery = "SELECT order_id FROM t";
+
+ Map<String, String> customProperties = new HashMap<>();
+ customProperties.put(MATERIALIZED_TABLE_DEFINITION_QUERY.key(), definitionQuery);
+ customProperties.put(MATERIALIZED_TABLE_INTERVAL_FRESHNESS.key(), "5");
+ customProperties.put(
+ MATERIALIZED_TABLE_INTERVAL_FRESHNESS_TIME_UNIT.key(),
+ IntervalFreshness.TimeUnit.SECOND.name());
+ customProperties.put(
+ MATERIALIZED_TABLE_LOGICAL_REFRESH_MODE.key(),
+ CatalogMaterializedTable.LogicalRefreshMode.CONTINUOUS.name());
+ customProperties.put(
+ MATERIALIZED_TABLE_REFRESH_MODE.key(),
+ CatalogMaterializedTable.RefreshMode.CONTINUOUS.name());
+ customProperties.put(
+ MATERIALIZED_TABLE_REFRESH_STATUS.key(),
+ CatalogMaterializedTable.RefreshStatus.INITIALIZING.name());
+ // Intentionally NO materialized-table.original-query / expanded-query entries —
+ // simulates a legacy persisted payload from before Flink 2.3.
+
+ org.apache.fluss.metadata.Schema flussSchema =
+ org.apache.fluss.metadata.Schema.newBuilder()
+ .column("order_id", org.apache.fluss.types.DataTypes.STRING())
+ .build();
+ TableDescriptor flussDescriptor =
+ TableDescriptor.builder()
+ .schema(flussSchema)
+ .distributedBy(1, Collections.singletonList("order_id"))
+ .customProperties(customProperties)
+ .build();
+
+ TablePath tablePath = TablePath.of("db", "table");
+ long currentMillis = System.currentTimeMillis();
+ TableInfo tableInfo =
+ TableInfo.of(
+ tablePath,
+ 1L,
+ 1,
+ flussDescriptor,
+ DEFAULT_REMOTE_DATA_DIR,
+ currentMillis,
+ currentMillis);
+
+ CatalogMaterializedTable flinkTable =
+ (CatalogMaterializedTable) FlinkConversions.toFlinkTable(tableInfo);
+
+ // Legacy payload: deserialization must succeed and definition-query must round-trip.
+ assertThat(flinkTable.getDefinitionQuery()).isEqualTo(definitionQuery);
+ // All three materialized-table query keys are consumed by the prefix-strip step,
+ // so they must not leak into getOptions().
+ assertThat(flinkTable.getOptions())
+ .doesNotContainKey(MATERIALIZED_TABLE_DEFINITION_QUERY.key())
+ .doesNotContainKey(MATERIALIZED_TABLE_ORIGINAL_QUERY.key())
+ .doesNotContainKey(MATERIALIZED_TABLE_EXPANDED_QUERY.key());
+ }
+
+ /**
+ * Verifies that {@link FlinkConversions#toFlussTable} writes the definition-query key (along
+ * with the rest of the materialized-table configuration) into {@code
+ * TableDescriptor.getCustomProperties()}, while keeping original/expanded-query out of the map
+ * when the active adapter does not expose them. Under the {@code fluss-flink-common} test
+ * classpath {@code CatalogMaterializedTableAdapter#getOriginalQuery} and {@code
+ * getExpandedQuery} return {@code null} for Flink 1.20/2.2, so those keys must be omitted
+ * entirely (a null value would otherwise break downstream readers that route customProperties
+ * through {@code Configuration.setString}). On Flink 2.3+ both keys would appear with their
+ * real values — that branch is covered by the dedicated test in the Flink 2.3 module.
+ */
+ @Test
+ void testMaterializedTableSerializesSeparateQueriesIntoCustomProperties() {
+ String definitionQuery = "SELECT order_id, orig_ts FROM t";
+
+ ResolvedSchema resolvedSchema =
+ new ResolvedSchema(
+ Arrays.asList(
+ Column.physical(
+ "order_id",
+ org.apache.flink.table.api.DataTypes.STRING().notNull()),
+ Column.physical(
+ "orig_ts",
+ org.apache.flink.table.api.DataTypes.TIMESTAMP())),
+ Collections.emptyList(),
+ UniqueConstraint.primaryKey(
+ "PK_order_id", Collections.singletonList("order_id")));
+ Map<String, String> flinkOptions = new HashMap<>();
+ flinkOptions.put("k1", "v1");
+
+ CatalogMaterializedTable flinkMaterializedTable =
+ CatalogMaterializedTable.newBuilder()
+ .schema(Schema.newBuilder().fromResolvedSchema(resolvedSchema).build())
+ .comment("test comment")
+ .options(flinkOptions)
+ .definitionQuery(definitionQuery)
+ .freshness(IntervalFreshness.of("5", IntervalFreshness.TimeUnit.SECOND))
+ .logicalRefreshMode(CatalogMaterializedTable.LogicalRefreshMode.CONTINUOUS)
+ .refreshMode(CatalogMaterializedTable.RefreshMode.CONTINUOUS)
+ .refreshStatus(CatalogMaterializedTable.RefreshStatus.INITIALIZING)
+ .build();
+
+ TableDescriptor flussTable =
+ FlinkConversions.toFlussTable(
+ new ResolvedCatalogMaterializedTable(
+ flinkMaterializedTable, resolvedSchema));
+
+ Map<String, String> customProperties = flussTable.getCustomProperties();
+
+ // definitionQuery is supported by every Flink version and is always persisted verbatim.
+ assertThat(customProperties)
+ .containsEntry(MATERIALIZED_TABLE_DEFINITION_QUERY.key(), definitionQuery);
+ // On Flink 1.20/2.2 the adapter returns null for original/expanded; the serialization
+ // must omit them entirely instead of writing a null entry, otherwise downstream readers
+ // (e.g. Configuration.setString) blow up with a NullPointerException.
+ assertThat(customProperties)
+ .doesNotContainKey(MATERIALIZED_TABLE_ORIGINAL_QUERY.key())
+ .doesNotContainKey(MATERIALIZED_TABLE_EXPANDED_QUERY.key());
+ }
+
/** Test refresh handler for testing purpose. */
public static class TestRefreshHandler implements RefreshHandler {
diff --git a/fluss-flink/pom.xml b/fluss-flink/pom.xml
index bd5726894..e059dcd41 100644
--- a/fluss-flink/pom.xml
+++ b/fluss-flink/pom.xml
@@ -37,6 +37,7 @@
<module>fluss-flink-1.19</module>
<module>fluss-flink-1.18</module>
<module>fluss-flink-2.2</module>
+ <module>fluss-flink-2.3</module>
<module>fluss-flink-tiering</module>
</modules>
diff --git a/fluss-test-coverage/pom.xml b/fluss-test-coverage/pom.xml
index 485dbf0ed..d8428c4a0 100644
--- a/fluss-test-coverage/pom.xml
+++ b/fluss-test-coverage/pom.xml
@@ -70,6 +70,13 @@
<scope>compile</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.fluss</groupId>
+ <artifactId>fluss-flink-2.3</artifactId>
+ <version>${project.version}</version>
+ <scope>compile</scope>
+ </dependency>
+
<dependency>
<groupId>org.apache.fluss</groupId>
<artifactId>fluss-flink-2.2</artifactId>
diff --git a/website/docs/engine-flink/delta-joins.mdx b/website/docs/engine-flink/delta-joins.mdx
index bc9bf7604..a6eb0f67d 100644
--- a/website/docs/engine-flink/delta-joins.mdx
+++ b/website/docs/engine-flink/delta-joins.mdx
@@ -194,9 +194,34 @@ There is a known issue ([FLINK-38399](https://issues.apache.org/jira/browse/FLIN
- All filters (including non-equi join conditions) must be applied on the upsert key.
- Filters, projections, and non-equi join conditions must not contain non-deterministic functions.
+### Flink 2.3
+
+#### Supported Features
+
+- Cascaded delta joins are now supported. Eligible join nodes in a query are sequentially converted into delta join nodes from source to sink (for example, `A JOIN B JOIN C` where applicable joins in the chain are optimized together).
+- Lookup joins are now supported after a delta join, allowing the result of a delta join to be further enriched via dimension tables.
+- Non-deterministic functions are now allowed in projections and filters placed between the delta join and its downstream operators (e.g., the sink).
+
+#### Limitations
+
+- The primary key or the prefix key of the tables must be included as part of the equivalence conditions in the join.
+- The join must be an INNER join. LEFT JOIN, RIGHT JOIN, and FULL OUTER JOIN are not supported.
+- The downstream node of the join must support idempotent updates, typically it's an upsert sink and should not have a `SinkUpsertMaterializer` node before it.
+ - Flink planner automatically inserts a `SinkUpsertMaterializer` when the sink's primary key does not fully cover the upstream update key.
+ - You can learn more details about `SinkUpsertMaterializer` by reading this [blog](https://www.ververica.com/blog/flink-sql-secrets-mastering-the-art-of-changelog-events).
+- Since delta join does not support to handle update-before messages, it is necessary to ensure that the entire pipeline can safely discard update-before messages. That means when consuming a CDC stream:
+ - The join key used in the delta join must be part of the upsert key.
+ - The sink's primary key must be the same as the upstream update key.
+ - All filters (including non-equi join conditions) must be applied on the upsert key.
+- Non-deterministic functions are not allowed in projections or filters between the source and the delta join.
+
+:::note
+Starting from Flink 2.3, the requirement that "the join key must be part of the primary key" has been relaxed to "the join key must be part of the upsert key". This paves the way for defining table-level **immutable columns** constraints in Fluss, which will allow additional columns to enrich the upsert key in a future release and enable delta join optimization in more scenarios.
+:::
+
## Future Plan
The Fluss and Flink community are actively working together on enhancing delta join. Planned improvements include:
- Support for additional join types, such as LEFT JOIN and FULL OUTER JOIN.
-- Support for multi-way joins involving more than two streams. This also requires Fluss to support defining multiple secondary index keys on a single table.
+- Support for **true multi-way joins** beyond the cascaded delta joins introduced in Flink 2.3. While Flink 2.3 can already optimize the `A JOIN B JOIN C` pattern by sequentially chaining delta joins, true multi-way join optimization (where a single join can be evaluated against multiple secondary indexes at once) still requires Fluss to support defining multiple secondary index keys on a single table.
- Support for more complex query patterns to optimize a wider range of join scenarios.
\ No newline at end of file
diff --git a/website/docs/engine-flink/getting-started.md b/website/docs/engine-flink/getting-started.md
index 16910fa90..7a4e6a760 100644
--- a/website/docs/engine-flink/getting-started.md
+++ b/website/docs/engine-flink/getting-started.md
@@ -15,6 +15,7 @@ Apache Fluss publishes the following JARs to Maven Central:
| Artifact | Jar |
|----------|-----|
+| Fluss connector for Flink 2.3 | [fluss-flink-2.3-$FLUSS_VERSION$.jar]($FLUSS_MAVEN_REPO_URL$/org/apache/fluss/fluss-flink-2.3/$FLUSS_VERSION$/fluss-flink-2.3-$FLUSS_VERSION$.jar) |
| Fluss connector for Flink 2.2 | [fluss-flink-2.2-$FLUSS_VERSION$.jar]($FLUSS_MAVEN_REPO_URL$/org/apache/fluss/fluss-flink-2.2/$FLUSS_VERSION$/fluss-flink-2.2-$FLUSS_VERSION$.jar) |
| Fluss connector for Flink 1.20 | [fluss-flink-1.20-$FLUSS_VERSION$.jar]($FLUSS_MAVEN_REPO_URL$/org/apache/fluss/fluss-flink-1.20/$FLUSS_VERSION$/fluss-flink-1.20-$FLUSS_VERSION$.jar) |
| Fluss connector for Flink 1.19 | [fluss-flink-1.19-$FLUSS_VERSION$.jar]($FLUSS_MAVEN_REPO_URL$/org/apache/fluss/fluss-flink-1.19/$FLUSS_VERSION$/fluss-flink-1.19-$FLUSS_VERSION$.jar) |
@@ -36,7 +37,7 @@ Verify downloaded JARs using the [verification instructions](/downloads#verifyin
## Supported Flink Versions
| Fluss Connector Versions | Supported Flink Versions |
|--------------------------|--------------------------|
-| $FLUSS_VERSION_SHORT$ | 1.18, 1.19, 1.20 |
+| $FLUSS_VERSION_SHORT$ | 1.18, 1.19, 1.20, 2.2, 2.3 |
## Feature Support
(fluss) 02/02: [flink] Fix stale delta join tests and add missing 2.3 coverage
Posted by ab...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
fresh-borzoni pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
commit bc04412ea961ce6a002c8852987e9edb0339ee09
Author: Anton Borisov <an...@fresha.com>
AuthorDate: Tue Aug 25 12:22:45 2026 +0200
[flink] Fix stale delta join tests and add missing 2.3 coverage
---
.../fluss/flink/catalog/Flink23CatalogTest.java | 18 ++-
.../fluss/flink/sink/Flink23SinkAdapterTest.java | 50 ++++++
.../fluss/flink/sink/Flink23TableSinkITCase.java | 1 +
.../fluss/flink/source/Flink23DeltaJoinITCase.java | 172 ++++++++++++++-------
4 files changed, 181 insertions(+), 60 deletions(-)
diff --git a/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/catalog/Flink23CatalogTest.java b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/catalog/Flink23CatalogTest.java
index d1cc974d5..4b89a9194 100644
--- a/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/catalog/Flink23CatalogTest.java
+++ b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/catalog/Flink23CatalogTest.java
@@ -142,7 +142,7 @@ public class Flink23CatalogTest extends FlinkCatalogTest {
@MultiVersionTest
void testMaterializedTableSerializesSeparateQueriesIntoCustomProperties() {
String originalQuery = "SELECT order_id, orig_ts FROM t";
- String expandedQuery = "SELECT defalut.t.order_id, defalut.t.orig_ts FROM defalut.t";
+ String expandedQuery = "SELECT default.t.order_id, default.t.orig_ts FROM default.t";
ResolvedSchema resolvedSchema =
new ResolvedSchema(
@@ -192,5 +192,21 @@ public class Flink23CatalogTest extends FlinkCatalogTest {
.containsEntry(
FlinkConnectorOptions.MATERIALIZED_TABLE_EXPANDED_QUERY.key(),
expandedQuery);
+
+ // read back: the two queries must stay distinct across the round trip
+ long currentMillis = System.currentTimeMillis();
+ CatalogMaterializedTable readBack =
+ (CatalogMaterializedTable)
+ FlinkConversions.toFlinkTable(
+ TableInfo.of(
+ TablePath.of("db", "table"),
+ 1L,
+ 1,
+ flussTable.withBucketCount(1),
+ DEFAULT_REMOTE_DATA_DIR,
+ currentMillis,
+ currentMillis));
+ assertThat(readBack.getOriginalQuery()).isEqualTo(originalQuery);
+ assertThat(readBack.getExpandedQuery()).isEqualTo(expandedQuery);
}
}
diff --git a/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/sink/Flink23SinkAdapterTest.java b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/sink/Flink23SinkAdapterTest.java
new file mode 100644
index 000000000..e21f77832
--- /dev/null
+++ b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/sink/Flink23SinkAdapterTest.java
@@ -0,0 +1,50 @@
+/*
+ * 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.fluss.flink.sink;
+
+import org.apache.fluss.flink.adapter.SinkAdapter;
+import org.apache.fluss.testutils.common.MultiVersionTest;
+
+import org.junit.jupiter.api.Test;
+
+import java.lang.reflect.Method;
+import java.lang.reflect.Modifier;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * {@link FlinkSink} is compiled against the common {@link SinkAdapter} but packaged against this
+ * module's one. A signature drift between them only surfaces as an {@code AbstractMethodError} when
+ * a writer is created at runtime, so assert the override resolves here.
+ */
+@MultiVersionTest
+class Flink23SinkAdapterTest {
+
+ @Test
+ void testFlinkSinkImplementsEveryAbstractSinkAdapterMethod() throws Exception {
+ assertThat(Modifier.isAbstract(FlinkSink.class.getModifiers())).isFalse();
+ for (Method method : SinkAdapter.class.getDeclaredMethods()) {
+ if (!Modifier.isAbstract(method.getModifiers())) {
+ continue;
+ }
+ Method override =
+ FlinkSink.class.getDeclaredMethod(method.getName(), method.getParameterTypes());
+ assertThat(Modifier.isAbstract(override.getModifiers())).isFalse();
+ }
+ }
+}
diff --git a/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/sink/Flink23TableSinkITCase.java b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/sink/Flink23TableSinkITCase.java
index 775f3689a..92d187100 100644
--- a/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/sink/Flink23TableSinkITCase.java
+++ b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/sink/Flink23TableSinkITCase.java
@@ -28,6 +28,7 @@ public class Flink23TableSinkITCase extends FlinkTableSinkITCase {
* FlinkChangelogModeInferenceProgram} does not reject plans whose inferred upsert key differs
* from the sink PK.
*/
+ @Override
protected void disableSinkRequireOnConflict() {
tEnv.getConfig().set(ExecutionConfigOptions.TABLE_EXEC_SINK_REQUIRE_ON_CONFLICT, false);
}
diff --git a/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/source/Flink23DeltaJoinITCase.java b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/source/Flink23DeltaJoinITCase.java
index b4f93193f..e45d87ff0 100644
--- a/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/source/Flink23DeltaJoinITCase.java
+++ b/fluss-flink/fluss-flink-2.3/src/test/java/org/apache/fluss/flink/source/Flink23DeltaJoinITCase.java
@@ -1014,88 +1014,142 @@ public class Flink23DeltaJoinITCase extends FlinkTestBase {
.hasMessageContaining("doesn't support to do delta join optimization");
}
- /**
- * Documents the SQL shape for a "lookup join after a delta join" pipeline (Flink 2.3): a delta
- * join wrapped in a subquery that adds a {@code PROCTIME()} time attribute, followed by a
- * downstream lookup (temporal) join against a Fluss dimension table. The proctime computed via
- * {@code PROCTIME()} in the inner subquery is consumed by the {@code FOR SYSTEM_TIME AS OF}
- * clause of the downstream lookup join.
- *
- * <p>Per Flink 2.3 "Supported Features and Limitations" docs: "Support for lookup join after a
- * delta join" and "Support for non-deterministic functions in projections and filters between
- * the delta join and downstream operators".
- *
- * <p>Today this is asserted as an expected failure to lock in the precise planner error that is
- * raised today: the {@code StreamPhysicalSnapshotOnCalcTableScanRule} in Flink 2.3 invokes
- * {@code FlussTableSource#getLookupRuntimeProvider} with an empty {@code
- * LookupContext.getKeys()} for the dim table in this scenario, and {@code
- * LookupNormalizer.validateAndCreateLookupNormalizer} then throws "Can't find expected key 'id'
- * in lookup keys []". The delta-join strategy is overridden to {@link
- * OptimizerConfigOptions.DeltaJoinStrategy#AUTO} so that the test is not blocked by {@code
- * StreamPhysicalDeltaJoinForceValidator}.
- */
@Test
- void testLookupJoinAfterDeltaJoin() {
- // allow the planner to choose; the validator only rejects when FORCE is set
- tEnv.getConfig()
- .set(
- OptimizerConfigOptions.TABLE_OPTIMIZER_DELTA_JOIN_STRATEGY,
- OptimizerConfigOptions.DeltaJoinStrategy.AUTO);
- String leftTableName = "left_table_lj_delta";
- String rightTableName = "right_table_lj_delta";
- String dimTableName = "dim_table_lj_delta";
- String sinkTableName = "sink_table_lj_delta";
+ void testCascadedDeltaJoin() {
+ String table1 = "cascade_t1";
+ String table2 = "cascade_t2";
+ String table3 = "cascade_t3";
+ String sinkTableName = "cascade_sink";
+
+ createSource(
+ table1,
+ "a1 int, c1 bigint, d1 int",
+ "c1, d1",
+ "c1",
+ null,
+ ImmutableMap.of("table.delete.behavior", "IGNORE"));
+ createSource(
+ table2,
+ "a2 int, c2 bigint, d2 int",
+ "c2, d2",
+ "c2",
+ null,
+ ImmutableMap.of("table.delete.behavior", "IGNORE"));
+ createSource(
+ table3,
+ "a3 int, c3 bigint, d3 int",
+ "c3, d3",
+ "c3",
+ null,
+ ImmutableMap.of("table.delete.behavior", "IGNORE"));
+ createSink(sinkTableName, "c1 bigint, d1 int, a1 int, a2 int, a3 int", "c1, d1");
+
+ String sql =
+ String.format(
+ "INSERT INTO %s SELECT t1.c1, t1.d1, t1.a1, t2.a2, t3.a3 FROM %s t1"
+ + " INNER JOIN %s t2 ON t1.c1 = t2.c2 AND t1.d1 = t2.d2"
+ + " INNER JOIN %s t3 ON t1.c1 = t3.c3 AND t1.d1 = t3.d3",
+ sinkTableName, table1, table2, table3);
+
+ // the outer delta join looks up through a multi-round chain over the inner one
+ assertThat(tEnv.explainSql(sql)).contains("round=[1]", "round=[2]");
+ }
+
+ @Test
+ void testDeltaJoinWithNonDeterministicFunctionBeforeSink() {
+ String leftTableName = "left_table_nondeterministic";
+ String rightTableName = "right_table_nondeterministic";
+ String sinkTableName = "sink_table_nondeterministic";
createSource(
leftTableName,
- "a1 int, b1 varchar, c1 bigint, d1 int, e1 bigint",
+ "a1 int, c1 bigint, d1 int",
"c1, d1",
"c1",
null,
ImmutableMap.of("table.delete.behavior", "IGNORE"));
createSource(
rightTableName,
- "a2 int, b2 varchar, c2 bigint, d2 int, e2 bigint",
+ "a2 int, c2 bigint, d2 int",
"c2, d2",
"c2",
null,
ImmutableMap.of("table.delete.behavior", "IGNORE"));
+ createSink(sinkTableName, "c1 bigint, d1 int, rand_val double", "c1, d1");
- tEnv.executeSql(
+ String sql =
String.format(
- "create table %s ("
- + " id int not null,"
- + " name varchar,"
- + " primary key (id) NOT ENFORCED"
- + ") with ("
- + " 'connector' = 'fluss',"
- + " 'bucket.num' = '1'"
- + ")",
- dimTableName));
+ "INSERT INTO %s SELECT l.c1, l.d1, RAND() FROM %s l INNER JOIN %s r"
+ + " ON l.c1 = r.c2 AND l.d1 = r.d2",
+ sinkTableName, leftTableName, rightTableName);
+
+ assertThat(tEnv.explainSql(sql)).contains("DeltaJoin(joinType=[InnerJoin]");
+ }
+ @Test
+ void testLookupJoinAfterDeltaJoin() throws Exception {
+ tEnv.getConfig().set(ExecutionConfigOptions.TABLE_EXEC_DELTA_JOIN_CACHE_ENABLED, false);
+ String leftTableName = "left_table_lookup_after_delta";
+ String rightTableName = "right_table_lookup_after_delta";
+ String dimTableName = "dim_table_lookup_after_delta";
+ String sinkTableName = "sink_table_lookup_after_delta";
+
+ createSource(
+ leftTableName,
+ "a1 int, c1 bigint, d1 int, ptime AS PROCTIME()",
+ "c1, d1",
+ "c1",
+ null,
+ ImmutableMap.of("table.delete.behavior", "IGNORE"));
+ createSource(
+ rightTableName,
+ "a2 int, c2 bigint, d2 int",
+ "c2, d2",
+ "c2",
+ null,
+ ImmutableMap.of("table.delete.behavior", "IGNORE"));
+ createSink(dimTableName, "id bigint, name varchar", "id");
createSink(
sinkTableName,
- "a1 int, b1 varchar, c1 bigint, d1 int, a2 int, dim_name varchar",
- "c1, d1");
+ "c1 bigint, d1 int, c2 bigint, d2 int, a2 int, name varchar",
+ "c1, d1, c2, d2");
+
+ TablePath leftTablePath = TablePath.of(DEFAULT_DB, leftTableName);
+ writeRows(conn, leftTablePath, Arrays.asList(row(1, 100L, 1), row(2, 200L, 2)), false);
+ FLUSS_CLUSTER_EXTENSION.triggerAndWaitSnapshot(leftTablePath);
+
+ TablePath rightTablePath = TablePath.of(DEFAULT_DB, rightTableName);
+ writeRows(conn, rightTablePath, Arrays.asList(row(10, 100L, 1), row(20, 200L, 2)), false);
+ FLUSS_CLUSTER_EXTENSION.triggerAndWaitSnapshot(rightTablePath);
+
+ TablePath dimTablePath = TablePath.of(DEFAULT_DB, dimTableName);
+ writeRows(
+ conn, dimTablePath, Arrays.asList(row(100L, "dim100"), row(200L, "dim200")), false);
+ // prefix join, so the sink is keyed by both sides' primary keys
String sql =
String.format(
- "INSERT INTO %s "
- + "SELECT tmp.a1, tmp.b1, tmp.c1, tmp.d1, tmp.a2, dim.name "
- + "FROM ("
- + " SELECT l.*, r.a2, PROCTIME() AS proc "
- + " FROM %s AS l INNER JOIN %s AS r ON l.c1 = r.c2"
- + ") tmp "
- + "JOIN %s FOR SYSTEM_TIME AS OF tmp.proc AS dim "
- + " ON dim.id = tmp.c1 "
- + "ON CONFLICT DO DEDUPLICATE",
+ "INSERT INTO %s SELECT l.c1, l.d1, r.c2, r.d2, r.a2, dim.name FROM %s l"
+ + " INNER JOIN %s r ON l.c1 = r.c2"
+ + " JOIN %s FOR SYSTEM_TIME AS OF l.ptime AS dim ON dim.id = l.c1",
sinkTableName, leftTableName, rightTableName, dimTableName);
- assertThatThrownBy(() -> tEnv.explainSql(sql))
- .hasRootCauseInstanceOf(org.apache.flink.table.api.TableException.class)
- .hasRootCauseMessage(
- "The Fluss lookup function supports lookup tables where the lookup keys"
- + " include all primary keys or all bucket keys. Can't find"
- + " expected key 'id' in lookup keys []");
+ String plan = tEnv.explainSql(sql);
+ assertThat(plan).contains("DeltaJoin(joinType=[InnerJoin]", "LookupJoin(");
+
+ tEnv.executeSql(sql);
+
+ CloseableIterator<Row> collected =
+ tEnv.executeSql(String.format("select * from %s", sinkTableName)).collect();
+ assertResultsIgnoreOrder(
+ collected,
+ Arrays.asList(
+ "+I[100, 1, 100, 1, 10, dim100]",
+ "-U[100, 1, 100, 1, 10, dim100]",
+ "+U[100, 1, 100, 1, 10, dim100]",
+ "+I[200, 2, 200, 2, 20, dim200]",
+ "-U[200, 2, 200, 2, 20, dim200]",
+ "+U[200, 2, 200, 2, 20, dim200]"),
+ true);
}
}