You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by yu...@apache.org on 2024/03/13 07:01:46 UTC
(incubator-paimon) branch master updated: [engine] Remove support for Flink 1.14 and Spark 2 (#2998)
This is an automated email from the ASF dual-hosted git repository.
yuzelin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 598ec08fc [engine] Remove support for Flink 1.14 and Spark 2 (#2998)
598ec08fc is described below
commit 598ec08fc96a46dcf5d3faacb5bf95a8b1eef78e
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Wed Mar 13 15:01:40 2024 +0800
[engine] Remove support for Flink 1.14 and Spark 2 (#2998)
---
.github/workflows/e2e-tests-1.14-jdk11.yml | 58 ----
.github/workflows/e2e-tests-1.14.yml | 58 ----
.github/workflows/utitcase-flink.yml | 2 +-
docs/content/engines/flink.md | 4 +-
docs/content/engines/overview.md | 2 +-
docs/content/engines/spark.md | 48 ---
docs/content/how-to/creating-tables.md | 10 -
docs/content/project/download.md | 4 -
paimon-e2e-tests/pom.xml | 11 -
.../java/org/apache/paimon/tests/HiveE2eTest.java | 2 -
.../org/apache/paimon/tests/LogStoreE2eTest.java | 2 -
.../java/org/apache/paimon/tests/SparkE2eTest.java | 4 +-
.../java/org/apache/paimon/tests/TypeE2eTest.java | 27 +-
paimon-flink/paimon-flink-1.14/pom.xml | 186 ------------
.../paimon/flink/PaimonDataStreamScanProvider.java | 49 ---
.../paimon/flink/PaimonDataStreamSinkProvider.java | 41 ---
.../org/apache/paimon/flink/TableConfigUtils.java | 30 --
.../flink/lookup/LookupRuntimeProviderFactory.java | 31 --
.../paimon/flink/lookup/OldLookupFunction.java | 52 ----
.../apache/paimon/flink/sink/FlinkTableSink.java | 39 ---
.../paimon/flink/source/table/RichTableSource.java | 53 ----
.../utils/SingleOutputStreamOperatorUtils.java | 36 ---
.../utils/StreamExecutionEnvironmentUtils.java | 48 ---
.../paimon/flink/utils/TableEnvironmentUtils.java | 60 ----
.../apache/paimon/flink/BatchFileStoreITCase.java | 180 -----------
.../org/apache/paimon/flink/CatalogITCaseBase.java | 147 ---------
.../paimon/flink/ContinuousFileStoreITCase.java | 332 ---------------------
.../org/apache/paimon/flink/FlinkActionITCase.java | 65 ----
.../org/apache/paimon/flink/LookupJoinITCase.java | 76 -----
.../org/apache/paimon/flink/WatermarkITCase.java | 81 -----
.../org.junit.jupiter.api.extension.Extension | 16 -
.../src/test/resources/log4j2-test.properties | 28 --
.../org/apache/paimon/flink/FlinkTableFactory.java | 26 --
.../paimon/flink/source/DataTableSource.java | 18 +-
paimon-flink/pom.xml | 2 -
paimon-spark/paimon-spark-2/pom.xml | 127 --------
.../apache/paimon/spark/SparkDataSourceReader.java | 141 ---------
.../apache/paimon/spark/SparkInputPartition.java | 64 ----
.../paimon/spark/SparkInputPartitionReader.java | 65 ----
.../java/org/apache/paimon/spark/SparkSource.java | 62 ----
...org.apache.spark.sql.sources.DataSourceRegister | 16 -
.../apache/paimon/spark/SimpleTableTestHelper.java | 84 ------
.../paimon/spark/SparkFilterConverterTest.java | 175 -----------
.../apache/paimon/spark/SparkInternalRowTest.java | 117 --------
.../org/apache/paimon/spark/SparkReadITCase.java | 133 ---------
.../org/apache/paimon/spark/SparkTypeTest.java | 96 ------
.../org.junit.jupiter.api.extension.Extension | 16 -
.../src/test/resources/log4j2-test.properties | 38 ---
.../org/apache/paimon/spark/SparkInternalRow.java | 74 ++++-
.../paimon/spark/SpecializedGettersReader.java | 109 -------
paimon-spark/pom.xml | 1 -
51 files changed, 90 insertions(+), 3056 deletions(-)
diff --git a/.github/workflows/e2e-tests-1.14-jdk11.yml b/.github/workflows/e2e-tests-1.14-jdk11.yml
deleted file mode 100644
index 2e6ff7f3c..000000000
--- a/.github/workflows/e2e-tests-1.14-jdk11.yml
+++ /dev/null
@@ -1,58 +0,0 @@
-################################################################################
-# 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.
-################################################################################
-
-name: End to End Tests Flink 1.14 on JDK 11
-
-on:
- issue_comment:
- types: [created, edited, deleted]
-
- # daily run
- schedule:
- - cron: "0 0 * * *"
-
-env:
- JDK_VERSION: 11
-
-jobs:
- build:
- if: |
- github.event_name == 'schedule' ||
- (contains(github.event.comment.html_url, '/pull/') && contains(github.event.comment.body, '/jdk11'))
- runs-on: ubuntu-latest
-
- steps:
- - name: Checkout code
- uses: actions/checkout@v2
- - name: Set up JDK ${{ env.JDK_VERSION }}
- uses: actions/setup-java@v2
- with:
- java-version: ${{ env.JDK_VERSION }}
- distribution: 'adopt'
- - name: Build Flink 1.14
- run: mvn -T 1C -B clean install -DskipTests -Pflink-1.14
- - name: Test Flink 1.14
- timeout-minutes: 60
- run: |
- # run tests with random timezone to find out timezone related bugs
- . .github/workflows/utils.sh
- jvm_timezone=$(random_timezone)
- echo "JVM timezone is set to $jvm_timezone"
- mvn -T 1C -B test -pl paimon-e2e-tests -Duser.timezone=$jvm_timezone -Pflink-1.14
- env:
- MAVEN_OPTS: -Xmx4096m
\ No newline at end of file
diff --git a/.github/workflows/e2e-tests-1.14.yml b/.github/workflows/e2e-tests-1.14.yml
deleted file mode 100644
index f53dbd8a7..000000000
--- a/.github/workflows/e2e-tests-1.14.yml
+++ /dev/null
@@ -1,58 +0,0 @@
-################################################################################
-# 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.
-################################################################################
-
-name: End to End Tests Flink 1.14
-
-on:
- push:
- pull_request:
- paths-ignore:
- - 'docs/**'
- - '**/*.md'
-
-env:
- JDK_VERSION: 8
-
-concurrency:
- group: ${{ github.workflow }}-${{ github.event_name }}-${{ github.event.number || github.run_id }}
- cancel-in-progress: true
-
-jobs:
- build:
- runs-on: ubuntu-latest
-
- steps:
- - name: Checkout code
- uses: actions/checkout@v2
- - name: Set up JDK ${{ env.JDK_VERSION }}
- uses: actions/setup-java@v2
- with:
- java-version: ${{ env.JDK_VERSION }}
- distribution: 'adopt'
- - name: Build Flink 1.14
- run: mvn -T 1C -B clean install -DskipTests -Pflink-1.14
- - name: Test Flink 1.14
- timeout-minutes: 60
- run: |
- # run tests with random timezone to find out timezone related bugs
- . .github/workflows/utils.sh
- jvm_timezone=$(random_timezone)
- echo "JVM timezone is set to $jvm_timezone"
- mvn -T 1C -B test -pl paimon-e2e-tests -Duser.timezone=$jvm_timezone -Pflink-1.14
- env:
- MAVEN_OPTS: -Xmx4096m
\ No newline at end of file
diff --git a/.github/workflows/utitcase-flink.yml b/.github/workflows/utitcase-flink.yml
index a5c83366f..b6cc9cbd1 100644
--- a/.github/workflows/utitcase-flink.yml
+++ b/.github/workflows/utitcase-flink.yml
@@ -52,7 +52,7 @@ jobs:
jvm_timezone=$(random_timezone)
echo "JVM timezone is set to $jvm_timezone"
test_modules=""
- for suffix in 1.14 1.15 1.16 1.17 1.18 common; do
+ for suffix in 1.15 1.16 1.17 1.18 common; do
test_modules+="org.apache.paimon:paimon-flink-${suffix},"
done
test_modules="${test_modules%,}"
diff --git a/docs/content/engines/flink.md b/docs/content/engines/flink.md
index 4ae48459c..eb5bd6836 100644
--- a/docs/content/engines/flink.md
+++ b/docs/content/engines/flink.md
@@ -30,7 +30,7 @@ This documentation is a guide for using Paimon in Flink.
## Preparing Paimon Jar File
-Paimon currently supports Flink 1.17, 1.16, 1.15 and 1.14. We recommend the latest Flink version for a better experience.
+Paimon currently supports Flink 1.18, 1.17, 1.16, 1.15. We recommend the latest Flink version for a better experience.
Download the jar file with corresponding version.
@@ -43,7 +43,6 @@ Download the jar file with corresponding version.
| Flink 1.17 | Bundled Jar | [paimon-flink-1.17-{{< version >}}.jar](https://repo.maven.apache.org/maven2/org/apache/paimon/paimon-flink-1.17/{{< version >}}/paimon-flink-1.17-{{< version >}}.jar) |
| Flink 1.16 | Bundled Jar | [paimon-flink-1.16-{{< version >}}.jar](https://repo.maven.apache.org/maven2/org/apache/paimon/paimon-flink-1.16/{{< version >}}/paimon-flink-1.16-{{< version >}}.jar) |
| Flink 1.15 | Bundled Jar | [paimon-flink-1.15-{{< version >}}.jar](https://repo.maven.apache.org/maven2/org/apache/paimon/paimon-flink-1.15/{{< version >}}/paimon-flink-1.15-{{< version >}}.jar) |
-| Flink 1.14 | Bundled Jar | [paimon-flink-1.14-{{< version >}}.jar](https://repo.maven.apache.org/maven2/org/apache/paimon/paimon-flink-1.14/{{< version >}}/paimon-flink-1.14-{{< version >}}.jar) |
| Flink Action | Action Jar | [paimon-flink-action-{{< version >}}.jar](https://repo.maven.apache.org/maven2/org/apache/paimon/paimon-flink-action/{{< version >}}/paimon-flink-action-{{< version >}}.jar) |
{{< /stable >}}
@@ -56,7 +55,6 @@ Download the jar file with corresponding version.
| Flink 1.17 | Bundled Jar | [paimon-flink-1.17-{{< version >}}.jar](https://repository.apache.org/snapshots/org/apache/paimon/paimon-flink-1.17/{{< version >}}/) |
| Flink 1.16 | Bundled Jar | [paimon-flink-1.16-{{< version >}}.jar](https://repository.apache.org/snapshots/org/apache/paimon/paimon-flink-1.16/{{< version >}}/) |
| Flink 1.15 | Bundled Jar | [paimon-flink-1.15-{{< version >}}.jar](https://repository.apache.org/snapshots/org/apache/paimon/paimon-flink-1.15/{{< version >}}/) |
-| Flink 1.14 | Bundled Jar | [paimon-flink-1.14-{{< version >}}.jar](https://repository.apache.org/snapshots/org/apache/paimon/paimon-flink-1.14/{{< version >}}/) |
| Flink Action | Action Jar | [paimon-flink-action-{{< version >}}.jar](https://repository.apache.org/snapshots/org/apache/paimon/paimon-flink-action/{{< version >}}/) |
{{< /unstable >}}
diff --git a/docs/content/engines/overview.md b/docs/content/engines/overview.md
index bff3d6658..05aa69e86 100644
--- a/docs/content/engines/overview.md
+++ b/docs/content/engines/overview.md
@@ -34,7 +34,7 @@ Apache Spark and Apache Hive.
| Engine | Version | Batch Read | Batch Write | Create Table | Alter Table | Streaming Write | Streaming Read | Batch Overwrite |
|:------:|:-------------:|:----------:|:-----------:|:------------:|:-----------:|:---------------:|:--------------:|:---------------:|
-| Flink | 1.14 - 1.18 | ✅ | ✅ | ✅ | ✅(1.17+) | ✅ | ✅ | ✅ |
+| Flink | 1.15 - 1.18 | ✅ | ✅ | ✅ | ✅(1.17+) | ✅ | ✅ | ✅ |
| Spark | 3.1 - 3.5 | ✅ | ✅ | ✅ | ✅ | ✅ | ✅(3.3+) | ✅ |
| Hive | 2.1 - 3.1 | ✅ | ✅ | ✅ | ❌ | ❌ | ❌ | ❌ |
| Spark | 2.4 | ✅ | ❌ | ❌ | ❌ | ❌ | ❌ | ❌ |
diff --git a/docs/content/engines/spark.md b/docs/content/engines/spark.md
index 680b198c8..ce0c45b39 100644
--- a/docs/content/engines/spark.md
+++ b/docs/content/engines/spark.md
@@ -724,51 +724,3 @@ All Spark's data types are available in package `org.apache.spark.sql.types`.
- Conversion between Spark's `UserDefinedType` and Paimon's `UserDefinedType` is not supported.
{{< /hint >}}
-
-## Spark 2
-
-Paimon supports Spark 2.4+. We highly recommend using versions above Spark3, as Spark2 only provides reading capabilities.
-
-{{< stable >}}
-
-Download [paimon-spark-2-{{< version >}}.jar](https://repo.maven.apache.org/maven2/org/apache/paimon/paimon-spark-2/{{< version >}}/paimon-spark-2-{{< version >}}.jar).
-
-{{< /stable >}}
-
-{{< unstable >}}
-
-Download [paimon-spark-2-{{< version >}}.jar](https://repository.apache.org/snapshots/org/apache/paimon/paimon-spark-2/{{< version >}}/).
-
-{{< /unstable >}}
-
-{{< hint info >}}
-
-If you are using HDFS, make sure that the environment variable `HADOOP_HOME` or `HADOOP_CONF_DIR` is set.
-
-{{< /hint >}}
-
-**Step 1: Prepare Test Data**
-
-Paimon currently only supports reading tables through Spark2. To create a Paimon table with records, please follow our [Flink quick start guide]({{< ref "engines/flink#quick-start" >}}).
-
-After the guide, all table files should be stored under the path `/tmp/paimon`, or the warehouse path you've specified.
-
-**Step 2: Specify Paimon Jar File**
-
-You can append path to paimon jar file to the `--jars` argument when starting `spark-shell`.
-
-```bash
-spark-shell ... --jars /path/to/paimon-spark-2-{{< version >}}.jar
-```
-
-Alternatively, you can copy `paimon-spark-2-{{< version >}}.jar` under `spark/jars` in your Spark installation directory.
-
-**Step 3: Query Table**
-
-Paimon with Spark 2.4 does not support DDL. You can use the `Dataset` reader and register the `Dataset` as a temporary table. In spark shell:
-
-```scala
-val dataset = spark.read.format("paimon").load("file:/tmp/paimon/default.db/word_count")
-dataset.createOrReplaceTempView("word_count")
-spark.sql("SELECT * FROM word_count").show()
-```
diff --git a/docs/content/how-to/creating-tables.md b/docs/content/how-to/creating-tables.md
index 4364d3077..bad9d418f 100644
--- a/docs/content/how-to/creating-tables.md
+++ b/docs/content/how-to/creating-tables.md
@@ -620,16 +620,6 @@ val dataset = spark.read.format("paimon").load("hdfs:///path/to/table")
{{< /tab >}}
-{{< tab "Spark2" >}}
-
-Spark2 only supports creating external tables through Scala API. The following Scala code loads the table located at `hdfs:///path/to/table` into a `DataSet`.
-
-```scala
-val dataset = spark.read.format("paimon").load("hdfs:///path/to/table")
-```
-
-{{< /tab >}}
-
{{< tab "Hive" >}}
To access existing paimon table, you can also register them as external tables in Hive. The following SQL creates an
diff --git a/docs/content/project/download.md b/docs/content/project/download.md
index e70f5fc76..4d8bc1fc9 100644
--- a/docs/content/project/download.md
+++ b/docs/content/project/download.md
@@ -38,14 +38,12 @@ This documentation is a guide for downloading Paimon Jars.
| Flink 1.17 | [paimon-flink-1.17-{{< version >}}.jar](https://repository.apache.org/snapshots/org/apache/paimon/paimon-flink-1.17/{{< version >}}/) |
| Flink 1.16 | [paimon-flink-1.16-{{< version >}}.jar](https://repository.apache.org/snapshots/org/apache/paimon/paimon-flink-1.16/{{< version >}}/) |
| Flink 1.15 | [paimon-flink-1.15-{{< version >}}.jar](https://repository.apache.org/snapshots/org/apache/paimon/paimon-flink-1.15/{{< version >}}/) |
-| Flink 1.14 | [paimon-flink-1.14-{{< version >}}.jar](https://repository.apache.org/snapshots/org/apache/paimon/paimon-flink-1.14/{{< version >}}/) |
| Flink Action | [paimon-flink-action-{{< version >}}.jar](https://repository.apache.org/snapshots/org/apache/paimon/paimon-flink-action/{{< version >}}/) |
| Spark 3.5 | [paimon-spark-3.5-{{< version >}}.jar](https://repository.apache.org/snapshots/org/apache/paimon/paimon-spark-3.5/{{< version >}}/) |
| Spark 3.4 | [paimon-spark-3.4-{{< version >}}.jar](https://repository.apache.org/snapshots/org/apache/paimon/paimon-spark-3.4/{{< version >}}/) |
| Spark 3.3 | [paimon-spark-3.3-{{< version >}}.jar](https://repository.apache.org/snapshots/org/apache/paimon/paimon-spark-3.3/{{< version >}}/) |
| Spark 3.2 | [paimon-spark-3.2-{{< version >}}.jar](https://repository.apache.org/snapshots/org/apache/paimon/paimon-spark-3.2/{{< version >}}/) |
| Spark 3.1 | [paimon-spark-3.1-{{< version >}}.jar](https://repository.apache.org/snapshots/org/apache/paimon/paimon-spark-3.1/{{< version >}}/) |
-| Spark 2 | [paimon-spark-2-{{< version >}}.jar](https://repository.apache.org/snapshots/org/apache/paimon/paimon-spark-2/{{< version >}}/) |
| Hive 3.1 | [paimon-hive-connector-3.1-{{< version >}}.jar](https://repository.apache.org/snapshots/org/apache/paimon/paimon-hive-connector-3.1/{{< version >}}/) |
| Hive 2.3 | [paimon-hive-connector-2.3-{{< version >}}.jar](https://repository.apache.org/snapshots/org/apache/paimon/paimon-hive-connector-2.3/{{< version >}}/) |
| Hive 2.2 | [paimon-hive-connector-2.2-{{< version >}}.jar](https://repository.apache.org/snapshots/org/apache/paimon/paimon-hive-connector-2.2/{{< version >}}/) |
@@ -73,14 +71,12 @@ This documentation is a guide for downloading Paimon Jars.
| Flink 1.17 | [paimon-flink-1.17-{{< version >}}.jar](https://repo.maven.apache.org/maven2/org/apache/paimon/paimon-flink-1.17/{{< version >}}/paimon-flink-1.17-{{< version >}}.jar) |
| Flink 1.16 | [paimon-flink-1.16-{{< version >}}.jar](https://repo.maven.apache.org/maven2/org/apache/paimon/paimon-flink-1.16/{{< version >}}/paimon-flink-1.16-{{< version >}}.jar) |
| Flink 1.15 | [paimon-flink-1.15-{{< version >}}.jar](https://repo.maven.apache.org/maven2/org/apache/paimon/paimon-flink-1.15/{{< version >}}/paimon-flink-1.15-{{< version >}}.jar) |
-| Flink 1.14 | [paimon-flink-1.14-{{< version >}}.jar](https://repo.maven.apache.org/maven2/org/apache/paimon/paimon-flink-1.14/{{< version >}}/paimon-flink-1.14-{{< version >}}.jar) |
| Flink Action | [paimon-flink-action-{{< version >}}.jar](https://repo.maven.apache.org/maven2/org/apache/paimon/paimon-flink-action/{{< version >}}/paimon-flink-action-{{< version >}}.jar) |
| Spark 3.5 | [paimon-spark-3.5-{{< version >}}.jar](https://repo.maven.apache.org/maven2/org/apache/paimon/paimon-spark-3.5/{{< version >}}/paimon-spark-3.5-{{< version >}}.jar) |
| Spark 3.4 | [paimon-spark-3.4-{{< version >}}.jar](https://repo.maven.apache.org/maven2/org/apache/paimon/paimon-spark-3.4/{{< version >}}/paimon-spark-3.4-{{< version >}}.jar) |
| Spark 3.3 | [paimon-spark-3.3-{{< version >}}.jar](https://repo.maven.apache.org/maven2/org/apache/paimon/paimon-spark-3.3/{{< version >}}/paimon-spark-3.3-{{< version >}}.jar) |
| Spark 3.2 | [paimon-spark-3.2-{{< version >}}.jar](https://repo.maven.apache.org/maven2/org/apache/paimon/paimon-spark-3.2/{{< version >}}/paimon-spark-3.2-{{< version >}}.jar) |
| Spark 3.1 | [paimon-spark-3.1-{{< version >}}.jar](https://repo.maven.apache.org/maven2/org/apache/paimon/paimon-spark-3.1/{{< version >}}/paimon-spark-3.1-{{< version >}}.jar) |
-| Spark 2 | [paimon-spark-2-{{< version >}}.jar](https://repo.maven.apache.org/maven2/org/apache/paimon/paimon-spark-2/{{< version >}}/paimon-spark-2-{{< version >}}.jar) |
| Hive 3.1 | [paimon-hive-connector-3.1-{{< version >}}.jar](https://repo.maven.apache.org/maven2/org/apache/paimon/paimon-hive-connector-3.1/{{< version >}}/paimon-hive-connector-3.1-{{< version >}}.jar) |
| Hive 2.3 | [paimon-hive-connector-2.3-{{< version >}}.jar](https://repo.maven.apache.org/maven2/org/apache/paimon/paimon-hive-connector-2.3/{{< version >}}/paimon-hive-connector-2.3-{{< version >}}.jar) |
| Hive 2.2 | [paimon-hive-connector-2.2-{{< version >}}.jar](https://repo.maven.apache.org/maven2/org/apache/paimon/paimon-hive-connector-2.2/{{< version >}}/paimon-hive-connector-2.2-{{< version >}}.jar) |
diff --git a/paimon-e2e-tests/pom.xml b/paimon-e2e-tests/pom.xml
index fbbcbcf77..604eb04c7 100644
--- a/paimon-e2e-tests/pom.xml
+++ b/paimon-e2e-tests/pom.xml
@@ -304,16 +304,5 @@ under the License.
<flink.sql.connector.hive>flink-sql-connector-hive-2.3.6_${scala.binary.version}</flink.sql.connector.hive>
</properties>
</profile>
-
- <profile>
- <id>flink-1.14</id>
- <properties>
- <test.flink.main.version>1.14</test.flink.main.version>
- <test.flink.version>1.14.6</test.flink.version>
- <test.flink.connector.kafka.version>${test.flink.version}</test.flink.connector.kafka.version>
- <flink.sql.connector.kafka>flink-sql-connector-kafka_${scala.binary.version}</flink.sql.connector.kafka>
- <flink.sql.connector.hive>flink-sql-connector-hive-2.3.6_${scala.binary.version}</flink.sql.connector.hive>
- </properties>
- </profile>
</profiles>
</project>
diff --git a/paimon-e2e-tests/src/test/java/org/apache/paimon/tests/HiveE2eTest.java b/paimon-e2e-tests/src/test/java/org/apache/paimon/tests/HiveE2eTest.java
index c431184a7..af5ef28bd 100644
--- a/paimon-e2e-tests/src/test/java/org/apache/paimon/tests/HiveE2eTest.java
+++ b/paimon-e2e-tests/src/test/java/org/apache/paimon/tests/HiveE2eTest.java
@@ -20,7 +20,6 @@ package org.apache.paimon.tests;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.condition.DisabledIfSystemProperty;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testcontainers.containers.Container;
@@ -36,7 +35,6 @@ import java.util.UUID;
* <p>NOTE: This test runs a complete Hadoop cluster in Docker, which requires a lot of memory. If
* you're running this test locally, make sure that the memory limit of your Docker is at least 8GB.
*/
-@DisabledIfSystemProperty(named = "test.flink.version", matches = "1.14.*")
public class HiveE2eTest extends E2eReaderTestBase {
private static final Logger LOG = LoggerFactory.getLogger(HiveE2eTest.class);
diff --git a/paimon-e2e-tests/src/test/java/org/apache/paimon/tests/LogStoreE2eTest.java b/paimon-e2e-tests/src/test/java/org/apache/paimon/tests/LogStoreE2eTest.java
index ec06bd2eb..7e099420a 100644
--- a/paimon-e2e-tests/src/test/java/org/apache/paimon/tests/LogStoreE2eTest.java
+++ b/paimon-e2e-tests/src/test/java/org/apache/paimon/tests/LogStoreE2eTest.java
@@ -20,12 +20,10 @@ package org.apache.paimon.tests;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.condition.DisabledIfSystemProperty;
import java.util.UUID;
/** Tests for reading and writing log store in stream jobs. */
-@DisabledIfSystemProperty(named = "test.flink.version", matches = "1.14.*")
public class LogStoreE2eTest extends E2eTestBase {
private String topicName;
diff --git a/paimon-e2e-tests/src/test/java/org/apache/paimon/tests/SparkE2eTest.java b/paimon-e2e-tests/src/test/java/org/apache/paimon/tests/SparkE2eTest.java
index b8e875216..5f2173af7 100644
--- a/paimon-e2e-tests/src/test/java/org/apache/paimon/tests/SparkE2eTest.java
+++ b/paimon-e2e-tests/src/test/java/org/apache/paimon/tests/SparkE2eTest.java
@@ -19,7 +19,6 @@
package org.apache.paimon.tests;
import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.condition.DisabledIfSystemProperty;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testcontainers.containers.Container;
@@ -30,7 +29,6 @@ import java.util.UUID;
import java.util.stream.Collectors;
/** Tests for reading paimon from Spark3. */
-@DisabledIfSystemProperty(named = "test.flink.version", matches = "1.14.*")
public class SparkE2eTest extends E2eReaderTestBase {
private static final Logger LOG = LoggerFactory.getLogger(SparkE2eTest.class);
@@ -41,7 +39,7 @@ public class SparkE2eTest extends E2eReaderTestBase {
@Test
public void testFlinkWriteAndSparkRead() throws Exception {
- String warehousePath = TEST_DATA_DIR + "/" + UUID.randomUUID().toString() + "_warehouse";
+ String warehousePath = TEST_DATA_DIR + "/" + UUID.randomUUID() + "_warehouse";
final String table = "T";
final String sparkTable = String.format("paimon.default.%s", table);
runSql(
diff --git a/paimon-e2e-tests/src/test/java/org/apache/paimon/tests/TypeE2eTest.java b/paimon-e2e-tests/src/test/java/org/apache/paimon/tests/TypeE2eTest.java
index dce208ded..1a6dcf236 100644
--- a/paimon-e2e-tests/src/test/java/org/apache/paimon/tests/TypeE2eTest.java
+++ b/paimon-e2e-tests/src/test/java/org/apache/paimon/tests/TypeE2eTest.java
@@ -97,16 +97,10 @@ public class TypeE2eTest extends E2eTestBase {
tableDdl,
createResultSink("result1", schema));
- String flinkVersion = System.getProperty("test.flink.version");
String expected =
- flinkVersion.startsWith("1.14")
- ? "true, 1, 10, 100, 1000, 1.1, 1.11, 12.456, "
- + "123456789123456789.12345678, hi, hello, table桌子store商店, "
- + "[116, 97, 98, 108, 101, -26, -95, -116, -27, -83, -112, 115, 116, 111, 114, 101, -27, -107, -122, -27, -70, -105], "
- + "2022-04-28, 2022-04-28T15:35:45.123, [hi, hello, null, test], +I[1, 10, 测试]"
- : "true, 1, 10, 100, 1000, 1.1, 1.11, 12.456, "
- + "123456789123456789.12345678, hi, hello, table桌子store商店, [116], "
- + "2022-04-28, 2022-04-28T15:35:45.123, [hi, hello, null, test], +I[1, 10, 测试]";
+ "true, 1, 10, 100, 1000, 1.1, 1.11, 12.456, "
+ + "123456789123456789.12345678, hi, hello, table桌子store商店, [116], "
+ + "2022-04-28, 2022-04-28T15:35:45.123, [hi, hello, null, test], +I[1, 10, 测试]";
checkResult(
expected,
"null, null, null, null, null, null, null, null, null, "
@@ -189,18 +183,11 @@ public class TypeE2eTest extends E2eTestBase {
tableDdl,
createResultSink("result1", schema));
- String flinkVersion = System.getProperty("test.flink.version");
String expected =
- flinkVersion.startsWith("1.14")
- ? "1, true, 1, 10, 100, 1000, 1.1, 1.11, 12.456, "
- + "123456789123456789.12345678, hi, hello, table桌子store商店, "
- + "[116, 97, 98, 108, 101, -26, -95, -116, -27, -83, -112, 115, 116, 111, 114, 101, -27, -107, -122, -27, -70, -105], "
- + "2022-04-28, 2022-04-28T15:35:45.123, [hi, hello, null, test], +I[1, 10, 测试], "
- + "{hi=1, test=3, hello=null}"
- : "1, true, 1, 10, 100, 1000, 1.1, 1.11, 12.456, "
- + "123456789123456789.12345678, hi, hello, table桌子store商店, [116], "
- + "2022-04-28, 2022-04-28T15:35:45.123, [hi, hello, null, test], +I[1, 10, 测试], "
- + "{hi=1, test=3, hello=null}";
+ "1, true, 1, 10, 100, 1000, 1.1, 1.11, 12.456, "
+ + "123456789123456789.12345678, hi, hello, table桌子store商店, [116], "
+ + "2022-04-28, 2022-04-28T15:35:45.123, [hi, hello, null, test], +I[1, 10, 测试], "
+ + "{hi=1, test=3, hello=null}";
checkResult(
expected,
"2, null, null, null, null, null, null, null, null, null, "
diff --git a/paimon-flink/paimon-flink-1.14/pom.xml b/paimon-flink/paimon-flink-1.14/pom.xml
deleted file mode 100644
index 0aeac707a..000000000
--- a/paimon-flink/paimon-flink-1.14/pom.xml
+++ /dev/null
@@ -1,186 +0,0 @@
-<?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>
- <artifactId>paimon-flink</artifactId>
- <groupId>org.apache.paimon</groupId>
- <version>0.8-SNAPSHOT</version>
- </parent>
-
- <artifactId>paimon-flink-1.14</artifactId>
- <name>Paimon : Flink : 1.14</name>
-
- <properties>
- <flink.version>1.14.6</flink.version>
- <frocksdbjni.version>6.20.3-ververica-2.0</frocksdbjni.version>
- </properties>
-
- <dependencies>
- <dependency>
- <groupId>org.apache.paimon</groupId>
- <artifactId>paimon-flink-common</artifactId>
- <version>${project.version}</version>
- <exclusions>
- <exclusion>
- <groupId>*</groupId>
- <artifactId>*</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
-
- <dependency>
- <groupId>org.apache.paimon</groupId>
- <artifactId>paimon-flink-cdc</artifactId>
- <version>${project.version}</version>
- <exclusions>
- <exclusion>
- <groupId>*</groupId>
- <artifactId>*</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
-
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
- <version>${flink.version}</version>
- <scope>provided</scope>
- </dependency>
-
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-table-api-java-bridge_${scala.binary.version}</artifactId>
- <version>${flink.version}</version>
- <scope>provided</scope>
- </dependency>
-
- <!-- test dependencies -->
-
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-test-utils_${scala.binary.version}</artifactId>
- <version>${flink.version}</version>
- <scope>test</scope>
- <exclusions>
- <exclusion>
- <groupId>log4j</groupId>
- <artifactId>log4j</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
-
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-table-planner_${scala.binary.version}</artifactId>
- <version>${flink.version}</version>
- <scope>test</scope>
- </dependency>
-
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-table-planner_${scala.binary.version}</artifactId>
- <version>${flink.version}</version>
- <type>test-jar</type>
- <scope>test</scope>
- </dependency>
-
- <dependency>
- <groupId>org.apache.hadoop</groupId>
- <artifactId>hadoop-hdfs-client</artifactId>
- <version>${hadoop.version}</version>
- <scope>provided</scope>
- </dependency>
-
- <dependency>
- <groupId>org.apache.hadoop</groupId>
- <artifactId>hadoop-mapreduce-client-core</artifactId>
- <version>${hadoop.version}</version>
- <scope>test</scope>
- <exclusions>
- <exclusion>
- <groupId>org.apache.curator</groupId>
- <artifactId>curator-test</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.apache.hadoop</groupId>
- <artifactId>hadoop-yarn-common</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.apache.avro</groupId>
- <artifactId>avro</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.codehaus.jackson</groupId>
- <artifactId>jackson-core-asl</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.slf4j</groupId>
- <artifactId>slf4j-log4j12</artifactId>
- </exclusion>
- <exclusion>
- <groupId>jdk.tools</groupId>
- <artifactId>jdk.tools</artifactId>
- </exclusion>
- <exclusion>
- <groupId>com.google.protobuf</groupId>
- <artifactId>protobuf-java</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
-
- <dependency>
- <groupId>com.ververica</groupId>
- <artifactId>frocksdbjni</artifactId>
- <version>${frocksdbjni.version}</version>
- <scope>test</scope>
- </dependency>
- </dependencies>
-
- <build>
- <plugins>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-shade-plugin</artifactId>
- <executions>
- <execution>
- <id>shade-paimon</id>
- <phase>package</phase>
- <goals>
- <goal>shade</goal>
- </goals>
- <configuration>
- <artifactSet>
- <includes combine.children="append">
- <include>org.apache.paimon:paimon-flink-common</include>
- <include>org.apache.paimon:paimon-flink-cdc</include>
- </includes>
- </artifactSet>
- </configuration>
- </execution>
- </executions>
- </plugin>
- </plugins>
- </build>
-</project>
-
diff --git a/paimon-flink/paimon-flink-1.14/src/main/java/org/apache/paimon/flink/PaimonDataStreamScanProvider.java b/paimon-flink/paimon-flink-1.14/src/main/java/org/apache/paimon/flink/PaimonDataStreamScanProvider.java
deleted file mode 100644
index b7bea8e17..000000000
--- a/paimon-flink/paimon-flink-1.14/src/main/java/org/apache/paimon/flink/PaimonDataStreamScanProvider.java
+++ /dev/null
@@ -1,49 +0,0 @@
-/*
- * 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.paimon.flink;
-
-import org.apache.flink.streaming.api.datastream.DataStream;
-import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.table.connector.source.DataStreamScanProvider;
-import org.apache.flink.table.data.RowData;
-
-import java.util.function.Function;
-
-/** Paimon {@link DataStreamScanProvider}. */
-public class PaimonDataStreamScanProvider implements DataStreamScanProvider {
-
- private final boolean isBounded;
- private final Function<StreamExecutionEnvironment, DataStream<RowData>> producer;
-
- public PaimonDataStreamScanProvider(
- boolean isBounded, Function<StreamExecutionEnvironment, DataStream<RowData>> producer) {
- this.isBounded = isBounded;
- this.producer = producer;
- }
-
- @Override
- public DataStream<RowData> produceDataStream(StreamExecutionEnvironment env) {
- return producer.apply(env);
- }
-
- @Override
- public boolean isBounded() {
- return isBounded;
- }
-}
diff --git a/paimon-flink/paimon-flink-1.14/src/main/java/org/apache/paimon/flink/PaimonDataStreamSinkProvider.java b/paimon-flink/paimon-flink-1.14/src/main/java/org/apache/paimon/flink/PaimonDataStreamSinkProvider.java
deleted file mode 100644
index 34aa8991d..000000000
--- a/paimon-flink/paimon-flink-1.14/src/main/java/org/apache/paimon/flink/PaimonDataStreamSinkProvider.java
+++ /dev/null
@@ -1,41 +0,0 @@
-/*
- * 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.paimon.flink;
-
-import org.apache.flink.streaming.api.datastream.DataStream;
-import org.apache.flink.streaming.api.datastream.DataStreamSink;
-import org.apache.flink.table.connector.sink.DataStreamSinkProvider;
-import org.apache.flink.table.data.RowData;
-
-import java.util.function.Function;
-
-/** Paimon {@link DataStreamSinkProvider}. */
-public class PaimonDataStreamSinkProvider implements DataStreamSinkProvider {
-
- private final Function<DataStream<RowData>, DataStreamSink<?>> producer;
-
- public PaimonDataStreamSinkProvider(Function<DataStream<RowData>, DataStreamSink<?>> producer) {
- this.producer = producer;
- }
-
- @Override
- public DataStreamSink<?> consumeDataStream(DataStream<RowData> dataStream) {
- return producer.apply(dataStream);
- }
-}
diff --git a/paimon-flink/paimon-flink-1.14/src/main/java/org/apache/paimon/flink/TableConfigUtils.java b/paimon-flink/paimon-flink-1.14/src/main/java/org/apache/paimon/flink/TableConfigUtils.java
deleted file mode 100644
index 1fef9f4a0..000000000
--- a/paimon-flink/paimon-flink-1.14/src/main/java/org/apache/paimon/flink/TableConfigUtils.java
+++ /dev/null
@@ -1,30 +0,0 @@
-/*
- * 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.paimon.flink;
-
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.ReadableConfig;
-
-/** Utils for {@link ReadableConfig}. */
-public class TableConfigUtils {
-
- public static Configuration extractConfiguration(ReadableConfig readableConfig) {
- return (Configuration) readableConfig;
- }
-}
diff --git a/paimon-flink/paimon-flink-1.14/src/main/java/org/apache/paimon/flink/lookup/LookupRuntimeProviderFactory.java b/paimon-flink/paimon-flink-1.14/src/main/java/org/apache/paimon/flink/lookup/LookupRuntimeProviderFactory.java
deleted file mode 100644
index 4221f4f1d..000000000
--- a/paimon-flink/paimon-flink-1.14/src/main/java/org/apache/paimon/flink/lookup/LookupRuntimeProviderFactory.java
+++ /dev/null
@@ -1,31 +0,0 @@
-/*
- * 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.paimon.flink.lookup;
-
-import org.apache.flink.table.connector.source.LookupTableSource.LookupRuntimeProvider;
-import org.apache.flink.table.connector.source.TableFunctionProvider;
-
-/** Factory to create {@link LookupRuntimeProvider}. */
-public class LookupRuntimeProviderFactory {
-
- public static LookupRuntimeProvider create(
- FileStoreLookupFunction function, boolean enableAsync, int asyncThreadNumber) {
- return TableFunctionProvider.of(new OldLookupFunction(function));
- }
-}
diff --git a/paimon-flink/paimon-flink-1.14/src/main/java/org/apache/paimon/flink/lookup/OldLookupFunction.java b/paimon-flink/paimon-flink-1.14/src/main/java/org/apache/paimon/flink/lookup/OldLookupFunction.java
deleted file mode 100644
index 402948935..000000000
--- a/paimon-flink/paimon-flink-1.14/src/main/java/org/apache/paimon/flink/lookup/OldLookupFunction.java
+++ /dev/null
@@ -1,52 +0,0 @@
-/*
- * 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.paimon.flink.lookup;
-
-import org.apache.flink.table.data.GenericRowData;
-import org.apache.flink.table.data.RowData;
-import org.apache.flink.table.functions.FunctionContext;
-import org.apache.flink.table.functions.TableFunction;
-
-/** Old lookup {@link TableFunction} for 1.15-. */
-public class OldLookupFunction extends TableFunction<RowData> {
-
- private static final long serialVersionUID = 1L;
-
- private final FileStoreLookupFunction function;
-
- public OldLookupFunction(FileStoreLookupFunction function) {
- this.function = function;
- }
-
- @Override
- public void open(FunctionContext context) throws Exception {
- function.open(context);
- }
-
- /** Used by code generation. */
- @SuppressWarnings("unused")
- public void eval(Object... values) {
- function.lookup(GenericRowData.of(values)).forEach(this::collect);
- }
-
- @Override
- public void close() throws Exception {
- function.close();
- }
-}
diff --git a/paimon-flink/paimon-flink-1.14/src/main/java/org/apache/paimon/flink/sink/FlinkTableSink.java b/paimon-flink/paimon-flink-1.14/src/main/java/org/apache/paimon/flink/sink/FlinkTableSink.java
deleted file mode 100644
index 054e48a20..000000000
--- a/paimon-flink/paimon-flink-1.14/src/main/java/org/apache/paimon/flink/sink/FlinkTableSink.java
+++ /dev/null
@@ -1,39 +0,0 @@
-/*
- * 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.paimon.flink.sink;
-
-import org.apache.paimon.flink.log.LogStoreTableFactory;
-import org.apache.paimon.table.Table;
-
-import org.apache.flink.table.catalog.ObjectIdentifier;
-import org.apache.flink.table.factories.DynamicTableFactory;
-
-import javax.annotation.Nullable;
-
-/** Table sink to create sink. */
-public class FlinkTableSink extends FlinkTableSinkBase {
-
- public FlinkTableSink(
- ObjectIdentifier tableIdentifier,
- Table table,
- DynamicTableFactory.Context context,
- @Nullable LogStoreTableFactory logStoreTableFactory) {
- super(tableIdentifier, table, context, logStoreTableFactory);
- }
-}
diff --git a/paimon-flink/paimon-flink-1.14/src/main/java/org/apache/paimon/flink/source/table/RichTableSource.java b/paimon-flink/paimon-flink-1.14/src/main/java/org/apache/paimon/flink/source/table/RichTableSource.java
deleted file mode 100644
index 50c8dc5ff..000000000
--- a/paimon-flink/paimon-flink-1.14/src/main/java/org/apache/paimon/flink/source/table/RichTableSource.java
+++ /dev/null
@@ -1,53 +0,0 @@
-/*
- * 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.paimon.flink.source.table;
-
-import org.apache.paimon.flink.source.FlinkTableSource;
-
-import org.apache.flink.api.common.eventtime.WatermarkStrategy;
-import org.apache.flink.table.connector.source.LookupTableSource;
-import org.apache.flink.table.connector.source.abilities.SupportsWatermarkPushDown;
-import org.apache.flink.table.data.RowData;
-
-/** The {@link BaseTableSource} with Lookup and watermark. */
-public class RichTableSource extends BaseTableSource
- implements LookupTableSource, SupportsWatermarkPushDown {
-
- private final FlinkTableSource source;
-
- public RichTableSource(FlinkTableSource source) {
- super(source);
- this.source = source;
- }
-
- @Override
- public RichTableSource copy() {
- return new RichTableSource(source.copy());
- }
-
- @Override
- public LookupRuntimeProvider getLookupRuntimeProvider(LookupContext context) {
- return source.getLookupRuntimeProvider(context);
- }
-
- @Override
- public void applyWatermark(WatermarkStrategy<RowData> watermarkStrategy) {
- source.pushWatermark(watermarkStrategy);
- }
-}
diff --git a/paimon-flink/paimon-flink-1.14/src/main/java/org/apache/paimon/flink/utils/SingleOutputStreamOperatorUtils.java b/paimon-flink/paimon-flink-1.14/src/main/java/org/apache/paimon/flink/utils/SingleOutputStreamOperatorUtils.java
deleted file mode 100644
index c7bf42a64..000000000
--- a/paimon-flink/paimon-flink-1.14/src/main/java/org/apache/paimon/flink/utils/SingleOutputStreamOperatorUtils.java
+++ /dev/null
@@ -1,36 +0,0 @@
-/*
- * 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.paimon.flink.utils;
-
-import org.apache.flink.streaming.api.datastream.DataStream;
-import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
-import org.apache.flink.util.OutputTag;
-
-/**
- * Utility methods for {@link SingleOutputStreamOperator}.
- *
- * <p>TODO: merge all Flink version related utils into one class.
- */
-public class SingleOutputStreamOperatorUtils {
-
- public static <T> DataStream<T> getSideOutput(
- SingleOutputStreamOperator<?> input, OutputTag<T> outputTag) {
- return input.getSideOutput(outputTag);
- }
-}
diff --git a/paimon-flink/paimon-flink-1.14/src/main/java/org/apache/paimon/flink/utils/StreamExecutionEnvironmentUtils.java b/paimon-flink/paimon-flink-1.14/src/main/java/org/apache/paimon/flink/utils/StreamExecutionEnvironmentUtils.java
deleted file mode 100644
index 05b8dcba0..000000000
--- a/paimon-flink/paimon-flink-1.14/src/main/java/org/apache/paimon/flink/utils/StreamExecutionEnvironmentUtils.java
+++ /dev/null
@@ -1,48 +0,0 @@
-/*
- * 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.paimon.flink.utils;
-
-import org.apache.flink.configuration.ReadableConfig;
-import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-
-import java.lang.reflect.Field;
-
-/** Utility methods for {@link StreamExecutionEnvironment}. */
-public class StreamExecutionEnvironmentUtils {
-
- public static ReadableConfig getConfiguration(StreamExecutionEnvironment env) {
- // For Flink 1.14 we have to use reflection to get configuration from execution environment.
- // See FLINK-26709 for more info.
- if (env.getClass()
- .getName()
- .equals("org.apache.flink.table.planner.utils.DummyStreamExecutionEnvironment")) {
- try {
- Field realExecEnvField = env.getClass().getDeclaredField("realExecEnv");
- realExecEnvField.setAccessible(true);
- env = (StreamExecutionEnvironment) realExecEnvField.get(env);
- } catch (NoSuchFieldException | IllegalAccessException e) {
- throw new RuntimeException(
- "Failed to get realExecEnv from DummyStreamExecutionEnvironment "
- + "by Java reflection. This is unexpected.",
- e);
- }
- }
- return env.getConfiguration();
- }
-}
diff --git a/paimon-flink/paimon-flink-1.14/src/main/java/org/apache/paimon/flink/utils/TableEnvironmentUtils.java b/paimon-flink/paimon-flink-1.14/src/main/java/org/apache/paimon/flink/utils/TableEnvironmentUtils.java
deleted file mode 100644
index d1f4c09b4..000000000
--- a/paimon-flink/paimon-flink-1.14/src/main/java/org/apache/paimon/flink/utils/TableEnvironmentUtils.java
+++ /dev/null
@@ -1,60 +0,0 @@
-/*
- * 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.paimon.flink.utils;
-
-import org.apache.flink.api.dag.Transformation;
-import org.apache.flink.table.api.TableEnvironment;
-import org.apache.flink.table.api.TableResult;
-import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
-
-import java.lang.reflect.InvocationTargetException;
-import java.lang.reflect.Method;
-import java.util.List;
-
-/** Utility methods for {@link TableEnvironment} and its subclasses. */
-public class TableEnvironmentUtils {
-
- /**
- * Invoke {@code TableEnvironmentImpl#executeInternal(List<Transformation<?>>, List<String>)}
- * from a {@link StreamTableEnvironment} instance through reflecting.
- */
- public static TableResult executeInternal(
- StreamTableEnvironment tEnv,
- List<Transformation<?>> transformations,
- List<String> sinkIdentifierNames) {
- Class<?> clazz = tEnv.getClass().getSuperclass();
- try {
- Method executeInternal =
- clazz.getDeclaredMethod("executeInternal", List.class, List.class);
- executeInternal.setAccessible(true);
-
- return (TableResult) executeInternal.invoke(tEnv, transformations, sinkIdentifierNames);
- } catch (NoSuchMethodException e) {
- throw new RuntimeException(
- "Failed to get 'TableEnvironmentImpl#executeInternal(List, List)' method "
- + "from given StreamTableEnvironment instance by Java reflection. This is unexpected.",
- e);
- } catch (IllegalAccessException | InvocationTargetException e) {
- throw new RuntimeException(
- "Failed to invoke 'TableEnvironmentImpl#executeInternal(List, List)' method "
- + "from given StreamTableEnvironment instance by Java reflection. This is unexpected.",
- e);
- }
- }
-}
diff --git a/paimon-flink/paimon-flink-1.14/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java b/paimon-flink/paimon-flink-1.14/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java
deleted file mode 100644
index 5d61204e5..000000000
--- a/paimon-flink/paimon-flink-1.14/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java
+++ /dev/null
@@ -1,180 +0,0 @@
-/*
- * 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.paimon.flink;
-
-import org.apache.paimon.CoreOptions;
-
-import org.apache.flink.types.Row;
-import org.junit.Test;
-
-import java.util.Collections;
-import java.util.List;
-
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
-
-/** ITCase for batch file store. */
-public class BatchFileStoreITCase extends CatalogITCaseBase {
-
- @Override
- protected List<String> ddl() {
- return Collections.singletonList("CREATE TABLE IF NOT EXISTS T (a INT, b INT, c INT)");
- }
-
- @Override
- protected int defaultParallelism() {
- return 1;
- }
-
- @Test
- public void testOverwriteEmpty() {
- batchSql("INSERT INTO T VALUES (1, 11, 111), (2, 22, 222)");
- assertThat(batchSql("SELECT * FROM T"))
- .containsExactlyInAnyOrder(Row.of(1, 11, 111), Row.of(2, 22, 222));
- batchSql("INSERT OVERWRITE T SELECT * FROM T WHERE 1 <> 1");
- assertThat(batchSql("SELECT * FROM T")).isEmpty();
- }
-
- @Test
- public void testTimeTravelRead() throws Exception {
- batchSql("INSERT INTO T VALUES (1, 11, 111), (2, 22, 222)");
- long time1 = System.currentTimeMillis();
-
- Thread.sleep(10);
- batchSql("INSERT INTO T VALUES (3, 33, 333), (4, 44, 444)");
- long time2 = System.currentTimeMillis();
-
- Thread.sleep(10);
- batchSql("INSERT INTO T VALUES (5, 55, 555), (6, 66, 666)");
- long time3 = System.currentTimeMillis();
-
- Thread.sleep(10);
- batchSql("INSERT INTO T VALUES (7, 77, 777), (8, 88, 888)");
-
- paimonTable("T").createTag("tag2", 2);
-
- assertThat(batchSql("SELECT * FROM T /*+ OPTIONS('scan.snapshot-id'='1') */"))
- .containsExactlyInAnyOrder(Row.of(1, 11, 111), Row.of(2, 22, 222));
- assertThat(
- batchSql(
- "SELECT * FROM T /*+ OPTIONS('scan.mode'='from-snapshot-full','scan.snapshot-id'='1') */"))
- .containsExactlyInAnyOrder(Row.of(1, 11, 111), Row.of(2, 22, 222));
-
- assertThat(batchSql("SELECT * FROM T /*+ OPTIONS('scan.snapshot-id'='0') */")).isEmpty();
- assertThat(
- batchSql(
- "SELECT * FROM T /*+ OPTIONS('scan.mode'='from-snapshot-full','scan.snapshot-id'='0') */"))
- .isEmpty();
-
- assertThat(
- batchSql(
- String.format(
- "SELECT * FROM T /*+ OPTIONS('scan.timestamp-millis'='%s') */",
- time1)))
- .containsExactlyInAnyOrder(Row.of(1, 11, 111), Row.of(2, 22, 222));
-
- assertThat(batchSql("SELECT * FROM T /*+ OPTIONS('scan.snapshot-id'='2') */"))
- .containsExactlyInAnyOrder(
- Row.of(1, 11, 111),
- Row.of(2, 22, 222),
- Row.of(3, 33, 333),
- Row.of(4, 44, 444));
- assertThat(
- batchSql(
- "SELECT * FROM T /*+ OPTIONS('scan.mode'='from-snapshot-full','scan.snapshot-id'='2') */"))
- .containsExactlyInAnyOrder(
- Row.of(1, 11, 111),
- Row.of(2, 22, 222),
- Row.of(3, 33, 333),
- Row.of(4, 44, 444));
- assertThat(
- batchSql(
- String.format(
- "SELECT * FROM T /*+ OPTIONS('scan.timestamp-millis'='%s') */",
- time2)))
- .containsExactlyInAnyOrder(
- Row.of(1, 11, 111),
- Row.of(2, 22, 222),
- Row.of(3, 33, 333),
- Row.of(4, 44, 444));
-
- assertThat(batchSql("SELECT * FROM T /*+ OPTIONS('scan.snapshot-id'='3') */"))
- .containsExactlyInAnyOrder(
- Row.of(1, 11, 111),
- Row.of(2, 22, 222),
- Row.of(3, 33, 333),
- Row.of(4, 44, 444),
- Row.of(5, 55, 555),
- Row.of(6, 66, 666));
- assertThat(
- batchSql(
- "SELECT * FROM T /*+ OPTIONS('scan.mode'='from-snapshot-full','scan.snapshot-id'='3') */"))
- .containsExactlyInAnyOrder(
- Row.of(1, 11, 111),
- Row.of(2, 22, 222),
- Row.of(3, 33, 333),
- Row.of(4, 44, 444),
- Row.of(5, 55, 555),
- Row.of(6, 66, 666));
- assertThat(
- batchSql(
- String.format(
- "SELECT * FROM T /*+ OPTIONS('scan.timestamp-millis'='%s') */",
- time3)))
- .containsExactlyInAnyOrder(
- Row.of(1, 11, 111),
- Row.of(2, 22, 222),
- Row.of(3, 33, 333),
- Row.of(4, 44, 444),
- Row.of(5, 55, 555),
- Row.of(6, 66, 666));
-
- assertThatThrownBy(
- () ->
- batchSql(
- String.format(
- "SELECT * FROM T /*+ OPTIONS('scan.timestamp-millis'='%s', 'scan.snapshot-id'='1') */",
- time3)))
- .hasRootCauseInstanceOf(IllegalArgumentException.class)
- .hasRootCauseMessage(
- "[scan.snapshot-id] must be null when you set [scan.timestamp-millis]");
-
- assertThatThrownBy(
- () ->
- batchSql(
- "SELECT * FROM T /*+ OPTIONS('scan.mode'='full', 'scan.snapshot-id'='1') */"))
- .hasRootCauseInstanceOf(IllegalArgumentException.class)
- .hasRootCauseMessage(
- "%s must be null when you use latest-full for scan.mode",
- CoreOptions.SCAN_SNAPSHOT_ID.key());
-
- // travel to tag
- assertThat(batchSql("SELECT * FROM T /*+ OPTIONS('scan.tag-name'='tag2') */"))
- .containsExactlyInAnyOrder(
- Row.of(1, 11, 111),
- Row.of(2, 22, 222),
- Row.of(3, 33, 333),
- Row.of(4, 44, 444));
-
- assertThatThrownBy(
- () -> batchSql("SELECT * FROM T /*+ OPTIONS('scan.tag-name'='unknown') */"))
- .hasRootCauseInstanceOf(IllegalArgumentException.class)
- .hasRootCauseMessage("Tag 'unknown' doesn't exist.");
- }
-}
diff --git a/paimon-flink/paimon-flink-1.14/src/test/java/org/apache/paimon/flink/CatalogITCaseBase.java b/paimon-flink/paimon-flink-1.14/src/test/java/org/apache/paimon/flink/CatalogITCaseBase.java
deleted file mode 100644
index d14e06495..000000000
--- a/paimon-flink/paimon-flink-1.14/src/test/java/org/apache/paimon/flink/CatalogITCaseBase.java
+++ /dev/null
@@ -1,147 +0,0 @@
-/*
- * 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.paimon.flink;
-
-import org.apache.paimon.Snapshot;
-import org.apache.paimon.catalog.Identifier;
-import org.apache.paimon.fs.Path;
-import org.apache.paimon.fs.local.LocalFileIO;
-import org.apache.paimon.table.Table;
-import org.apache.paimon.utils.SnapshotManager;
-
-import org.apache.paimon.shade.guava30.com.google.common.collect.ImmutableList;
-
-import org.apache.flink.table.api.EnvironmentSettings;
-import org.apache.flink.table.api.TableEnvironment;
-import org.apache.flink.table.api.config.ExecutionConfigOptions;
-import org.apache.flink.table.api.internal.TableEnvironmentImpl;
-import org.apache.flink.table.delegation.Parser;
-import org.apache.flink.table.operations.Operation;
-import org.apache.flink.table.operations.ddl.CreateCatalogOperation;
-import org.apache.flink.test.util.AbstractTestBase;
-import org.apache.flink.types.Row;
-import org.apache.flink.util.CloseableIterator;
-import org.junit.Before;
-
-import javax.annotation.Nullable;
-
-import java.io.File;
-import java.io.IOException;
-import java.time.Duration;
-import java.util.Collections;
-import java.util.List;
-
-import static org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions.CHECKPOINTING_INTERVAL;
-
-/** ITCase for catalog. */
-public abstract class CatalogITCaseBase extends AbstractTestBase {
-
- protected TableEnvironment tEnv;
- protected TableEnvironment sEnv;
- protected String path;
-
- @Before
- public void before() throws IOException {
- tEnv = TableEnvironment.create(EnvironmentSettings.newInstance().inBatchMode().build());
- String catalog = "PAIMON";
- path = getTempDirPath("paimon");
- tEnv.executeSql(
- String.format(
- "CREATE CATALOG %s WITH (" + "'type'='paimon', 'warehouse'='%s')",
- catalog, path));
- tEnv.useCatalog(catalog);
-
- sEnv = TableEnvironment.create(EnvironmentSettings.newInstance().inStreamingMode().build());
- sEnv.getConfig().getConfiguration().set(CHECKPOINTING_INTERVAL, Duration.ofMillis(100));
- sEnv.registerCatalog(catalog, tEnv.getCatalog(catalog).get());
- sEnv.useCatalog(catalog);
-
- setParallelism(defaultParallelism());
- prepareEnv();
- }
-
- private void prepareEnv() {
- Parser parser = ((TableEnvironmentImpl) tEnv).getParser();
- for (String ddl : ddl()) {
- tEnv.executeSql(ddl);
- List<Operation> operations = parser.parse(ddl);
- if (operations.size() == 1) {
- Operation operation = operations.get(0);
- if (operation instanceof CreateCatalogOperation) {
- String name = ((CreateCatalogOperation) operation).getCatalogName();
- sEnv.registerCatalog(name, tEnv.getCatalog(name).orElse(null));
- }
- }
- }
- }
-
- protected void setParallelism(int parallelism) {
- tEnv.getConfig()
- .getConfiguration()
- .set(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM, parallelism);
- sEnv.getConfig()
- .getConfiguration()
- .set(ExecutionConfigOptions.TABLE_EXEC_RESOURCE_DEFAULT_PARALLELISM, parallelism);
- }
-
- protected int defaultParallelism() {
- return 2;
- }
-
- protected List<String> ddl() {
- return Collections.emptyList();
- }
-
- protected List<Row> batchSql(String query, Object... args) {
- return sql(query, args);
- }
-
- protected List<Row> sql(String query, Object... args) {
- try (CloseableIterator<Row> iter = tEnv.executeSql(String.format(query, args)).collect()) {
- return ImmutableList.copyOf(iter);
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- }
-
- protected CloseableIterator<Row> streamSqlIter(String query, Object... args) {
- return sEnv.executeSql(String.format(query, args)).collect();
- }
-
- protected Table paimonTable(String tableName)
- throws org.apache.paimon.catalog.Catalog.TableNotExistException {
- FlinkCatalog flinkCatalog = (FlinkCatalog) tEnv.getCatalog(tEnv.getCurrentCatalog()).get();
- org.apache.paimon.catalog.Catalog catalog = flinkCatalog.catalog();
- return catalog.getTable(Identifier.create(tEnv.getCurrentDatabase(), tableName));
- }
-
- protected Path getTableDirectory(String tableName) {
- return new Path(
- new File(path, String.format("%s.db/%s", tEnv.getCurrentDatabase(), tableName))
- .toString());
- }
-
- @Nullable
- protected Snapshot findLatestSnapshot(String tableName) {
- SnapshotManager snapshotManager =
- new SnapshotManager(LocalFileIO.create(), getTableDirectory(tableName));
- Long id = snapshotManager.latestSnapshotId();
- return id == null ? null : snapshotManager.snapshot(id);
- }
-}
diff --git a/paimon-flink/paimon-flink-1.14/src/test/java/org/apache/paimon/flink/ContinuousFileStoreITCase.java b/paimon-flink/paimon-flink-1.14/src/test/java/org/apache/paimon/flink/ContinuousFileStoreITCase.java
deleted file mode 100644
index 0e14aaeeb..000000000
--- a/paimon-flink/paimon-flink-1.14/src/test/java/org/apache/paimon/flink/ContinuousFileStoreITCase.java
+++ /dev/null
@@ -1,332 +0,0 @@
-/*
- * 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.paimon.flink;
-
-import org.apache.paimon.Snapshot;
-import org.apache.paimon.fs.local.LocalFileIO;
-import org.apache.paimon.utils.BlockingIterator;
-import org.apache.paimon.utils.SnapshotManager;
-
-import org.apache.paimon.shade.guava30.com.google.common.collect.ImmutableList;
-
-import org.apache.flink.types.Row;
-import org.junit.Test;
-
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Comparator;
-import java.util.List;
-
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
-
-/** SQL ITCase for continuous file store. */
-public class ContinuousFileStoreITCase extends CatalogITCaseBase {
-
- @Override
- protected List<String> ddl() {
- return Arrays.asList(
- "CREATE TABLE IF NOT EXISTS T1 (a STRING, b STRING, c STRING) WITH ('bucket' = '1')",
- "CREATE TABLE IF NOT EXISTS T2 (a STRING, b STRING, c STRING, PRIMARY KEY (a) NOT ENFORCED)"
- + " WITH ('changelog-producer'='input', 'bucket' = '1')");
- }
-
- @Test
- public void testWithoutPrimaryKey() throws Exception {
- testSimple("T1");
- }
-
- @Test
- public void testWithPrimaryKey() throws Exception {
- testSimple("T2");
- }
-
- @Test
- public void testProjectionWithoutPrimaryKey() throws Exception {
- testProjection("T1");
- }
-
- @Test
- public void testProjectionWithPrimaryKey() throws Exception {
- testProjection("T2");
- }
-
- private void testSimple(String table) throws Exception {
- BlockingIterator<Row, Row> iterator =
- BlockingIterator.of(streamSqlIter("SELECT * FROM %s", table));
-
- batchSql("INSERT INTO %s VALUES ('1', '2', '3'), ('4', '5', '6')", table);
- assertThat(iterator.collect(2))
- .containsExactlyInAnyOrder(Row.of("1", "2", "3"), Row.of("4", "5", "6"));
-
- batchSql("INSERT INTO %s VALUES ('7', '8', '9')", table);
- assertThat(iterator.collect(1)).containsExactlyInAnyOrder(Row.of("7", "8", "9"));
- iterator.close();
- }
-
- private void testProjection(String table) throws Exception {
- BlockingIterator<Row, Row> iterator =
- BlockingIterator.of(streamSqlIter("SELECT b, c FROM %s", table));
-
- batchSql("INSERT INTO %s VALUES ('1', '2', '3'), ('4', '5', '6')", table);
- assertThat(iterator.collect(2))
- .containsExactlyInAnyOrder(Row.of("2", "3"), Row.of("5", "6"));
-
- batchSql("INSERT INTO %s VALUES ('7', '8', '9')", table);
- assertThat(iterator.collect(1)).containsExactlyInAnyOrder(Row.of("8", "9"));
- iterator.close();
- }
-
- @Test
- public void testContinuousLatest() throws Exception {
- batchSql("INSERT INTO T1 VALUES ('1', '2', '3'), ('4', '5', '6')");
-
- BlockingIterator<Row, Row> iterator =
- BlockingIterator.of(
- streamSqlIter("SELECT * FROM T1 /*+ OPTIONS('log.scan'='latest') */"));
-
- batchSql("INSERT INTO T1 VALUES ('7', '8', '9'), ('10', '11', '12')");
- assertThat(iterator.collect(2))
- .containsExactlyInAnyOrder(Row.of("7", "8", "9"), Row.of("10", "11", "12"));
- iterator.close();
- }
-
- @Test
- public void testContinuousFromTimestamp() throws Exception {
- String sql =
- "SELECT * FROM T1 /*+ OPTIONS('log.scan'='from-timestamp', 'log.scan.timestamp-millis'='%s') */";
-
- // empty table
- BlockingIterator<Row, Row> iterator = BlockingIterator.of(streamSqlIter(sql, 0));
- batchSql("INSERT INTO T1 VALUES ('1', '2', '3'), ('4', '5', '6')");
- batchSql("INSERT INTO T1 VALUES ('7', '8', '9'), ('10', '11', '12')");
- assertThat(iterator.collect(2))
- .containsExactlyInAnyOrder(Row.of("1", "2", "3"), Row.of("4", "5", "6"));
- iterator.close();
-
- SnapshotManager snapshotManager =
- new SnapshotManager(LocalFileIO.create(), getTableDirectory("T1"));
- List<Snapshot> snapshots =
- new ArrayList<>(ImmutableList.copyOf(snapshotManager.snapshots()));
- snapshots.sort(Comparator.comparingLong(Snapshot::timeMillis));
- Snapshot first = snapshots.get(0);
- Snapshot second = snapshots.get(1);
-
- // before second snapshot
- iterator = BlockingIterator.of(streamSqlIter(sql, second.timeMillis() - 1));
- batchSql("INSERT INTO T1 VALUES ('13', '14', '15')");
- assertThat(iterator.collect(3))
- .containsExactlyInAnyOrder(
- Row.of("7", "8", "9"), Row.of("10", "11", "12"), Row.of("13", "14", "15"));
- iterator.close();
-
- // from second snapshot
- iterator = BlockingIterator.of(streamSqlIter(sql, second.timeMillis()));
- assertThat(iterator.collect(3))
- .containsExactlyInAnyOrder(
- Row.of("7", "8", "9"), Row.of("10", "11", "12"), Row.of("13", "14", "15"));
- iterator.close();
-
- // from start
- iterator = BlockingIterator.of(streamSqlIter(sql, first.timeMillis() - 1));
- assertThat(iterator.collect(5))
- .containsExactlyInAnyOrder(
- Row.of("1", "2", "3"),
- Row.of("4", "5", "6"),
- Row.of("7", "8", "9"),
- Row.of("10", "11", "12"),
- Row.of("13", "14", "15"));
- iterator.close();
-
- // from end
- iterator =
- BlockingIterator.of(
- streamSqlIter(
- sql,
- snapshotManager
- .snapshot(snapshotManager.latestSnapshotId())
- .timeMillis()
- + 1));
- batchSql("INSERT INTO T1 VALUES ('16', '17', '18')");
- assertThat(iterator.collect(1)).containsExactlyInAnyOrder(Row.of("16", "17", "18"));
- iterator.close();
- }
-
- @Test
- public void testLackStartupTimestamp() {
- assertThatThrownBy(
- () ->
- streamSqlIter(
- "SELECT * FROM T1 /*+ OPTIONS('log.scan'='from-timestamp') */"))
- .hasMessageContaining("Unable to create a source for reading table");
- }
-
- @Test
- public void testConfigureStartupTimestamp() throws Exception {
- // Configure 'log.scan.timestamp-millis' without 'log.scan'.
- BlockingIterator<Row, Row> iterator =
- BlockingIterator.of(
- streamSqlIter(
- "SELECT * FROM T1 /*+ OPTIONS('log.scan.timestamp-millis'='%s') */",
- 0));
- batchSql("INSERT INTO T1 VALUES ('1', '2', '3'), ('4', '5', '6')");
- batchSql("INSERT INTO T1 VALUES ('7', '8', '9'), ('10', '11', '12')");
- assertThat(iterator.collect(2))
- .containsExactlyInAnyOrder(Row.of("1", "2", "3"), Row.of("4", "5", "6"));
- iterator.close();
-
- // Configure 'log.scan.timestamp-millis' with 'log.scan=latest'.
- assertThatThrownBy(
- () ->
- streamSqlIter(
- "SELECT * FROM T1 /*+ OPTIONS('log.scan'='latest', 'log.scan.timestamp-millis'='%s') */",
- 0))
- .hasMessageContaining("Unable to create a source for reading table");
- }
-
- @Test
- public void testConfigureStartupSnapshot() throws Exception {
- // Configure 'scan.snapshot-id' without 'scan.mode'.
- batchSql("INSERT INTO T1 VALUES ('1', '2', '3'), ('4', '5', '6')");
- batchSql("INSERT INTO T1 VALUES ('7', '8', '9'), ('10', '11', '12')");
- BlockingIterator<Row, Row> iterator =
- BlockingIterator.of(
- streamSqlIter(
- "SELECT * FROM T1 /*+ OPTIONS('scan.snapshot-id'='%s') */", 1));
- assertThat(iterator.collect(2))
- .containsExactlyInAnyOrder(Row.of("1", "2", "3"), Row.of("4", "5", "6"));
- iterator.close();
-
- // Start from earliest snapshot
- iterator =
- BlockingIterator.of(
- streamSqlIter(
- "SELECT * FROM T1 /*+ OPTIONS('scan.snapshot-id'='%s') */", 0));
- assertThat(iterator.collect(2))
- .containsExactlyInAnyOrder(Row.of("1", "2", "3"), Row.of("4", "5", "6"));
- iterator.close();
-
- iterator =
- BlockingIterator.of(
- streamSqlIter(
- "SELECT * FROM T1 /*+ OPTIONS('scan.snapshot-id'='%s') */", 1));
- assertThat(iterator.collect(2))
- .containsExactlyInAnyOrder(Row.of("1", "2", "3"), Row.of("4", "5", "6"));
- iterator.close();
-
- iterator =
- BlockingIterator.of(
- streamSqlIter(
- "SELECT * FROM T1 /*+ OPTIONS('scan.snapshot-id'='%s') */", 2));
- assertThat(iterator.collect(2))
- .containsExactlyInAnyOrder(Row.of("7", "8", "9"), (Row.of("10", "11", "12")));
- iterator.close();
-
- // Configure 'scan.snapshot-id' with 'scan.mode=latest'.
- assertThatThrownBy(
- () ->
- streamSqlIter(
- "SELECT * FROM T1 /*+ OPTIONS('scan.mode'='latest', 'scan.snapshot-id'='%s') */",
- 0))
- .hasCauseInstanceOf(IllegalArgumentException.class)
- .hasRootCauseMessage(
- "scan.snapshot-id must be null when you use latest for scan.mode");
- }
-
- @Test
- public void testConfigureStartupSnapshotFull() throws Exception {
- // Configure 'scan.snapshot-id' with 'scan.mode'='from-snapshot-full'.
- batchSql("INSERT INTO T1 VALUES ('1', '2', '3'), ('4', '5', '6')");
- batchSql("INSERT INTO T1 VALUES ('7', '8', '9'), ('10', '11', '12')");
-
- // Start from earliest snapshot
- BlockingIterator<Row, Row> iterator =
- BlockingIterator.of(
- streamSqlIter(
- "SELECT * FROM T1 /*+ OPTIONS('scan.mode'='from-snapshot-full','scan.snapshot-id'='%s') */",
- 1));
- assertThat(iterator.collect(2))
- .containsExactlyInAnyOrder(Row.of("1", "2", "3"), Row.of("4", "5", "6"));
- iterator.close();
-
- batchSql("INSERT INTO T1 VALUES ('13', '14', '15')");
-
- iterator =
- BlockingIterator.of(
- streamSqlIter(
- "SELECT * FROM T1 /*+ OPTIONS('scan.mode'='from-snapshot-full','scan.snapshot-id'='%s') */",
- 2));
- assertThat(iterator.collect(4))
- .containsExactlyInAnyOrder(
- Row.of("1", "2", "3"),
- Row.of("4", "5", "6"),
- Row.of("7", "8", "9"),
- Row.of("10", "11", "12"));
- iterator.close();
-
- iterator =
- BlockingIterator.of(
- streamSqlIter(
- "SELECT * FROM T1 /*+ OPTIONS('scan.mode'='from-snapshot-full','scan.snapshot-id'='%s') */",
- 2));
- assertThat(iterator.collect(5))
- .containsExactlyInAnyOrder(
- Row.of("1", "2", "3"),
- Row.of("4", "5", "6"),
- Row.of("7", "8", "9"),
- Row.of("10", "11", "12"),
- Row.of("13", "14", "15"));
- iterator.close();
- }
-
- @Test
- public void testIgnoreOverwrite() throws Exception {
- BlockingIterator<Row, Row> iterator =
- BlockingIterator.of(streamSqlIter("SELECT * FROM T1"));
-
- batchSql("INSERT INTO T1 VALUES ('1', '2', '3'), ('4', '5', '6')");
- assertThat(iterator.collect(2))
- .containsExactlyInAnyOrder(Row.of("1", "2", "3"), Row.of("4", "5", "6"));
-
- // should ignore this overwrite
- batchSql("INSERT OVERWRITE T1 VALUES ('7', '8', '9')");
-
- batchSql("INSERT INTO T1 VALUES ('9', '10', '11')");
- assertThat(iterator.collect(1)).containsExactlyInAnyOrder(Row.of("9", "10", "11"));
- iterator.close();
- }
-
- @Test
- public void testUnsupportedUpsert() {
- assertThatThrownBy(
- () ->
- streamSqlIter(
- "SELECT * FROM T1 /*+ OPTIONS('log.changelog-mode'='upsert') */"),
- "File store continuous reading does not support upsert changelog mode");
- }
-
- @Test
- public void testUnsupportedEventual() {
- assertThatThrownBy(
- () ->
- streamSqlIter(
- "SELECT * FROM T1 /*+ OPTIONS('log.consistency'='eventual') */"),
- "File store continuous reading does not support eventual consistency mode");
- }
-}
diff --git a/paimon-flink/paimon-flink-1.14/src/test/java/org/apache/paimon/flink/FlinkActionITCase.java b/paimon-flink/paimon-flink-1.14/src/test/java/org/apache/paimon/flink/FlinkActionITCase.java
deleted file mode 100644
index 22df2a57a..000000000
--- a/paimon-flink/paimon-flink-1.14/src/test/java/org/apache/paimon/flink/FlinkActionITCase.java
+++ /dev/null
@@ -1,65 +0,0 @@
-/*
- * 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.paimon.flink;
-
-import org.apache.paimon.flink.action.DeleteAction;
-import org.apache.paimon.utils.BlockingIterator;
-
-import org.apache.flink.types.Row;
-import org.junit.Test;
-
-import java.util.Collections;
-import java.util.List;
-
-import static org.apache.flink.table.planner.factories.TestValuesTableFactory.changelogRow;
-import static org.assertj.core.api.Assertions.assertThat;
-
-/** ITCase for Flink action. */
-public class FlinkActionITCase extends CatalogITCaseBase {
-
- protected int defaultParallelism() {
- return 1;
- }
-
- protected List<String> ddl() {
- return Collections.singletonList(
- "CREATE TABLE T (k INT, v STRING, PRIMARY KEY (k) NOT ENFORCED)");
- }
-
- @Test
- public void testDeleteAction() throws Exception {
- batchSql("INSERT INTO T VALUES (1, 'Hi'), (2, 'Hello'), (3, 'World')");
-
- DeleteAction action =
- new DeleteAction(path, "default", "T", "k = 1", Collections.emptyMap());
-
- BlockingIterator<Row, Row> iterator =
- BlockingIterator.of(sEnv.executeSql("SELECT * FROM T").collect());
-
- action.run();
-
- assertThat(iterator.collect(4))
- .containsExactlyInAnyOrder(
- changelogRow("+I", 1, "Hi"),
- changelogRow("+I", 2, "Hello"),
- changelogRow("+I", 3, "World"),
- changelogRow("-D", 1, "Hi"));
- iterator.close();
- }
-}
diff --git a/paimon-flink/paimon-flink-1.14/src/test/java/org/apache/paimon/flink/LookupJoinITCase.java b/paimon-flink/paimon-flink-1.14/src/test/java/org/apache/paimon/flink/LookupJoinITCase.java
deleted file mode 100644
index f9f2a10b6..000000000
--- a/paimon-flink/paimon-flink-1.14/src/test/java/org/apache/paimon/flink/LookupJoinITCase.java
+++ /dev/null
@@ -1,76 +0,0 @@
-/*
- * 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.paimon.flink;
-
-import org.apache.paimon.utils.BlockingIterator;
-
-import org.apache.flink.types.Row;
-import org.junit.Test;
-
-import java.util.Arrays;
-import java.util.List;
-
-import static org.assertj.core.api.Assertions.assertThat;
-
-/** ITCase for lookup join. */
-public class LookupJoinITCase extends CatalogITCaseBase {
-
- @Override
- public List<String> ddl() {
- return Arrays.asList(
- "CREATE TABLE T (i INT, `proctime` AS PROCTIME())",
- "CREATE TABLE DIM (i INT PRIMARY KEY NOT ENFORCED, j INT, k1 INT, k2 INT) WITH"
- + " ('continuous.discovery-interval'='1 ms')");
- }
-
- @Override
- protected int defaultParallelism() {
- return 1;
- }
-
- @Test
- public void testLookup() throws Exception {
- batchSql("INSERT INTO DIM VALUES (1, 11, 111, 1111), (2, 22, 222, 2222)");
-
- String query =
- "SELECT T.i, D.j, D.k1, D.k2 FROM T LEFT JOIN DIM for system_time as of T.proctime AS D ON T.i = D.i";
- BlockingIterator<Row, Row> iterator = BlockingIterator.of(sEnv.executeSql(query).collect());
-
- batchSql("INSERT INTO T VALUES (1), (2), (3)");
- List<Row> result = iterator.collect(3);
- assertThat(result)
- .containsExactlyInAnyOrder(
- Row.of(1, 11, 111, 1111),
- Row.of(2, 22, 222, 2222),
- Row.of(3, null, null, null));
-
- batchSql("INSERT INTO DIM VALUES (2, 44, 444, 4444), (3, 33, 333, 3333)");
- Thread.sleep(2000); // wait refresh
- batchSql("INSERT INTO T VALUES (1), (2), (3), (4)");
- result = iterator.collect(4);
- assertThat(result)
- .containsExactlyInAnyOrder(
- Row.of(1, 11, 111, 1111),
- Row.of(2, 44, 444, 4444),
- Row.of(3, 33, 333, 3333),
- Row.of(4, null, null, null));
-
- iterator.close();
- }
-}
diff --git a/paimon-flink/paimon-flink-1.14/src/test/java/org/apache/paimon/flink/WatermarkITCase.java b/paimon-flink/paimon-flink-1.14/src/test/java/org/apache/paimon/flink/WatermarkITCase.java
deleted file mode 100644
index 2d50a1dd9..000000000
--- a/paimon-flink/paimon-flink-1.14/src/test/java/org/apache/paimon/flink/WatermarkITCase.java
+++ /dev/null
@@ -1,81 +0,0 @@
-/*
- * 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.paimon.flink;
-
-import org.apache.paimon.utils.BlockingIterator;
-
-import org.apache.flink.types.Row;
-import org.junit.Test;
-
-import java.time.LocalDateTime;
-
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
-
-/** SQL ITCase for watermark definition. */
-public class WatermarkITCase extends CatalogITCaseBase {
-
- @Override
- protected int defaultParallelism() {
- return 1;
- }
-
- @Test
- public void testWatermark() throws Exception {
- innerTestWatermark();
- }
-
- @Test
- public void testWatermarkAlignment() {
- assertThatThrownBy(
- () ->
- innerTestWatermark(
- "'scan.watermark.idle-timeout'='1s'",
- "'scan.watermark.alignment.group'='group'",
- "'scan.watermark.alignment.update-interval'='2s'",
- "'scan.watermark.alignment.max-drift'='1s'"))
- .hasMessageContaining(
- "Flink 1.14 does not support watermark alignment, please check your Flink version");
- }
-
- private void innerTestWatermark(String... options) throws Exception {
- sql(
- "CREATE TABLE T (f0 INT, ts TIMESTAMP(3), WATERMARK FOR ts AS ts) WITH ("
- + String.join(",", options)
- + ")");
-
- BlockingIterator<Row, Row> select =
- BlockingIterator.of(
- streamSqlIter(
- "SELECT window_start, window_end, SUM(f0) FROM TABLE("
- + "TUMBLE(TABLE T, DESCRIPTOR(ts), INTERVAL '10' MINUTES))\n"
- + " GROUP BY window_start, window_end"));
-
- sql("INSERT INTO T VALUES (1, TIMESTAMP '2023-02-02 12:00:00')");
- sql("INSERT INTO T VALUES (1, TIMESTAMP '2023-02-02 12:10:01')");
-
- assertThat(select.collect(1))
- .containsExactlyInAnyOrder(
- Row.of(
- LocalDateTime.parse("2023-02-02T12:00"),
- LocalDateTime.parse("2023-02-02T12:10"),
- 1));
- select.close();
- }
-}
diff --git a/paimon-flink/paimon-flink-1.14/src/test/resources/META-INF/services/org.junit.jupiter.api.extension.Extension b/paimon-flink/paimon-flink-1.14/src/test/resources/META-INF/services/org.junit.jupiter.api.extension.Extension
deleted file mode 100644
index 82214a9c7..000000000
--- a/paimon-flink/paimon-flink-1.14/src/test/resources/META-INF/services/org.junit.jupiter.api.extension.Extension
+++ /dev/null
@@ -1,16 +0,0 @@
-# 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.paimon.testutils.junit.TestLoggerExtension
diff --git a/paimon-flink/paimon-flink-1.14/src/test/resources/log4j2-test.properties b/paimon-flink/paimon-flink-1.14/src/test/resources/log4j2-test.properties
deleted file mode 100644
index 1b3980d15..000000000
--- a/paimon-flink/paimon-flink-1.14/src/test/resources/log4j2-test.properties
+++ /dev/null
@@ -1,28 +0,0 @@
-################################################################################
-# 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 [%tid %t] %-5p %c %x - %m%n
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkTableFactory.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkTableFactory.java
index a79897c9d..4c9f8ff9b 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkTableFactory.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkTableFactory.java
@@ -28,7 +28,6 @@ import org.apache.flink.table.catalog.ResolvedCatalogTable;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.factories.DynamicTableFactory;
-import org.apache.flink.table.factories.FactoryUtil;
import static org.apache.paimon.CoreOptions.AUTO_CREATE;
import static org.apache.paimon.flink.FlinkCatalogFactory.IDENTIFIER;
@@ -43,32 +42,12 @@ public class FlinkTableFactory extends AbstractFlinkTableFactory {
@Override
public DynamicTableSource createDynamicTableSource(Context context) {
- if (isFlinkTable(context)) {
- // only Flink 1.14 temporary table will come here
- return FactoryUtil.createTableSource(
- null,
- context.getObjectIdentifier(),
- context.getCatalogTable(),
- context.getConfiguration(),
- context.getClassLoader(),
- context.isTemporary());
- }
createTableIfNeeded(context);
return super.createDynamicTableSource(context);
}
@Override
public DynamicTableSink createDynamicTableSink(Context context) {
- if (isFlinkTable(context)) {
- // only Flink 1.14 temporary table will come here
- return FactoryUtil.createTableSink(
- null,
- context.getObjectIdentifier(),
- context.getCatalogTable(),
- context.getConfiguration(),
- context.getClassLoader(),
- context.isTemporary());
- }
createTableIfNeeded(context);
return super.createDynamicTableSink(context);
}
@@ -90,9 +69,4 @@ public class FlinkTableFactory extends AbstractFlinkTableFactory {
}
}
}
-
- private boolean isFlinkTable(Context context) {
- String identifier = context.getCatalogTable().getOptions().get(FactoryUtil.CONNECTOR.key());
- return identifier != null && !IDENTIFIER.equals(identifier);
- }
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/DataTableSource.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/DataTableSource.java
index dd3783deb..fd29dd048 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/DataTableSource.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/DataTableSource.java
@@ -191,18 +191,12 @@ public class DataTableSource extends FlinkTableSource {
}
String watermarkAlignGroup = options.get(SCAN_WATERMARK_ALIGNMENT_GROUP);
if (watermarkAlignGroup != null) {
- try {
- watermarkStrategy =
- WatermarkAlignUtils.withWatermarkAlignment(
- watermarkStrategy,
- watermarkAlignGroup,
- options.get(SCAN_WATERMARK_ALIGNMENT_MAX_DRIFT),
- options.get(SCAN_WATERMARK_ALIGNMENT_UPDATE_INTERVAL));
- } catch (NoSuchMethodError error) {
- throw new RuntimeException(
- "Flink 1.14 does not support watermark alignment, please check your Flink version.",
- error);
- }
+ watermarkStrategy =
+ WatermarkAlignUtils.withWatermarkAlignment(
+ watermarkStrategy,
+ watermarkAlignGroup,
+ options.get(SCAN_WATERMARK_ALIGNMENT_MAX_DRIFT),
+ options.get(SCAN_WATERMARK_ALIGNMENT_UPDATE_INTERVAL));
}
}
diff --git a/paimon-flink/pom.xml b/paimon-flink/pom.xml
index df6ee6acb..e81d4be87 100644
--- a/paimon-flink/pom.xml
+++ b/paimon-flink/pom.xml
@@ -35,7 +35,6 @@ under the License.
<modules>
<module>paimon-flink-common</module>
- <module>paimon-flink-1.14</module>
<module>paimon-flink-1.15</module>
<module>paimon-flink-1.16</module>
<module>paimon-flink-1.17</module>
@@ -135,7 +134,6 @@ under the License.
<id>skip-paimon-flink-tests</id>
<modules>
<module>paimon-flink-common</module>
- <module>paimon-flink-1.14</module>
<module>paimon-flink-1.15</module>
<module>paimon-flink-1.16</module>
<module>paimon-flink-1.17</module>
diff --git a/paimon-spark/paimon-spark-2/pom.xml b/paimon-spark/paimon-spark-2/pom.xml
deleted file mode 100644
index ccba0a812..000000000
--- a/paimon-spark/paimon-spark-2/pom.xml
+++ /dev/null
@@ -1,127 +0,0 @@
-<?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.paimon</groupId>
- <artifactId>paimon-spark</artifactId>
- <version>0.8-SNAPSHOT</version>
- </parent>
-
- <artifactId>paimon-spark-2</artifactId>
- <name>Paimon : Spark : 2</name>
-
- <properties>
- <spark2.version>2.4.8</spark2.version>
- <jackson.version>2.13.3</jackson.version>
- <asm6.version>4.10</asm6.version>
- </properties>
-
- <dependencies>
- <dependency>
- <groupId>org.apache.paimon</groupId>
- <artifactId>paimon-spark-common</artifactId>
- <version>${project.version}</version>
- <exclusions>
- <exclusion>
- <groupId>*</groupId>
- <artifactId>*</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
-
- <dependency>
- <groupId>org.apache.spark</groupId>
- <artifactId>spark-sql_2.11</artifactId>
- <version>${spark2.version}</version>
- <exclusions>
- <exclusion>
- <groupId>org.apache.avro</groupId>
- <artifactId>avro</artifactId>
- </exclusion>
- <!-- The Jackson dependencies of Spark 2.4.8 have vulnerabilities. -->
- <exclusion>
- <groupId>com.fasterxml.jackson.core</groupId>
- <artifactId>*</artifactId>
- </exclusion>
- <exclusion>
- <groupId>com.fasterxml.jackson.module</groupId>
- <artifactId>*</artifactId>
- </exclusion>
- <exclusion>
- <groupId>log4j</groupId>
- <artifactId>log4j</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.slf4j</groupId>
- <artifactId>slf4j-log4j12</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.apache.xbean</groupId>
- <artifactId>xbean-asm6-shaded</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.glassfish.jersey.media</groupId>
- <artifactId>jersey-media-json-jackson</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
-
- <dependency>
- <groupId>com.fasterxml.jackson.module</groupId>
- <artifactId>jackson-module-scala_2.11</artifactId>
- <version>${jackson.version}</version>
- <scope>test</scope>
- </dependency>
- <dependency>
- <groupId>org.apache.xbean</groupId>
- <artifactId>xbean-asm6-shaded</artifactId>
- <version>${asm6.version}</version>
- </dependency>
- </dependencies>
-
- <build>
- <plugins>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-shade-plugin</artifactId>
- <executions>
- <execution>
- <id>shade-paimon</id>
- <phase>package</phase>
- <goals>
- <goal>shade</goal>
- </goals>
- <configuration>
- <artifactSet>
- <includes combine.children="append">
- <include>org.apache.paimon:paimon-spark-common</include>
- </includes>
- </artifactSet>
- </configuration>
- </execution>
- </executions>
- </plugin>
- </plugins>
- </build>
-</project>
diff --git a/paimon-spark/paimon-spark-2/src/main/java/org/apache/paimon/spark/SparkDataSourceReader.java b/paimon-spark/paimon-spark-2/src/main/java/org/apache/paimon/spark/SparkDataSourceReader.java
deleted file mode 100644
index 8dae36a5b..000000000
--- a/paimon-spark/paimon-spark-2/src/main/java/org/apache/paimon/spark/SparkDataSourceReader.java
+++ /dev/null
@@ -1,141 +0,0 @@
-/*
- * 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.paimon.spark;
-
-import org.apache.paimon.predicate.Predicate;
-import org.apache.paimon.table.Table;
-import org.apache.paimon.table.source.ReadBuilder;
-import org.apache.paimon.table.source.Split;
-import org.apache.paimon.types.RowType;
-import org.apache.paimon.utils.TypeUtils;
-
-import org.apache.spark.sql.catalyst.InternalRow;
-import org.apache.spark.sql.sources.Filter;
-import org.apache.spark.sql.sources.v2.reader.DataSourceReader;
-import org.apache.spark.sql.sources.v2.reader.InputPartition;
-import org.apache.spark.sql.sources.v2.reader.Statistics;
-import org.apache.spark.sql.sources.v2.reader.SupportsPushDownFilters;
-import org.apache.spark.sql.sources.v2.reader.SupportsPushDownRequiredColumns;
-import org.apache.spark.sql.sources.v2.reader.SupportsReportStatistics;
-import org.apache.spark.sql.types.StructType;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.OptionalLong;
-import java.util.stream.Collectors;
-
-/** A Spark {@link DataSourceReader} for paimon. */
-public class SparkDataSourceReader
- implements SupportsPushDownFilters,
- SupportsPushDownRequiredColumns,
- SupportsReportStatistics {
-
- private final Table table;
-
- private List<Predicate> predicates = new ArrayList<>();
- private Filter[] pushedFilters;
- private int[] projectedFields;
- private List<Split> splits;
-
- public SparkDataSourceReader(Table table) {
- this.table = table;
- }
-
- @Override
- public Filter[] pushFilters(Filter[] filters) {
- SparkFilterConverter converter = new SparkFilterConverter(table.rowType());
- List<Predicate> predicates = new ArrayList<>();
- List<Filter> pushed = new ArrayList<>();
- for (Filter filter : filters) {
- try {
- predicates.add(converter.convert(filter));
- pushed.add(filter);
- } catch (UnsupportedOperationException ignore) {
- }
- }
- this.predicates = predicates;
- this.pushedFilters = pushed.toArray(new Filter[0]);
- return filters;
- }
-
- @Override
- public Filter[] pushedFilters() {
- return pushedFilters;
- }
-
- @Override
- public void pruneColumns(StructType requiredSchema) {
- String[] pruneFields = requiredSchema.fieldNames();
- List<String> fieldNames = table.rowType().getFieldNames();
- int[] projected = new int[pruneFields.length];
- for (int i = 0; i < projected.length; i++) {
- projected[i] = fieldNames.indexOf(pruneFields[i]);
- }
- this.projectedFields = projected;
- }
-
- @Override
- public Statistics estimateStatistics() {
- long rowCount = 0L;
-
- for (Split split : splits()) {
- rowCount += split.rowCount();
- }
-
- final long numRows = rowCount;
- final long sizeInBytes = readSchema().defaultSize() * numRows;
-
- return new Statistics() {
- @Override
- public OptionalLong sizeInBytes() {
- return OptionalLong.of(sizeInBytes);
- }
-
- @Override
- public OptionalLong numRows() {
- return OptionalLong.of(numRows);
- }
- };
- }
-
- @Override
- public StructType readSchema() {
- RowType rowType = table.rowType();
- return SparkTypeUtils.fromPaimonRowType(
- projectedFields == null ? rowType : TypeUtils.project(rowType, projectedFields));
- }
-
- private ReadBuilder readBuilder() {
- return table.newReadBuilder().withFilter(predicates).withProjection(projectedFields);
- }
-
- @Override
- public List<InputPartition<InternalRow>> planInputPartitions() {
- return splits().stream()
- .map(split -> new SparkInputPartition(readBuilder(), split))
- .collect(Collectors.toList());
- }
-
- protected List<Split> splits() {
- if (splits == null) {
- this.splits = readBuilder().newScan().plan().splits();
- }
- return splits;
- }
-}
diff --git a/paimon-spark/paimon-spark-2/src/main/java/org/apache/paimon/spark/SparkInputPartition.java b/paimon-spark/paimon-spark-2/src/main/java/org/apache/paimon/spark/SparkInputPartition.java
deleted file mode 100644
index 71bb61b00..000000000
--- a/paimon-spark/paimon-spark-2/src/main/java/org/apache/paimon/spark/SparkInputPartition.java
+++ /dev/null
@@ -1,64 +0,0 @@
-/*
- * 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.paimon.spark;
-
-import org.apache.paimon.data.InternalRow;
-import org.apache.paimon.reader.RecordReader;
-import org.apache.paimon.reader.RecordReaderIterator;
-import org.apache.paimon.table.source.ReadBuilder;
-import org.apache.paimon.table.source.Split;
-
-import org.apache.spark.sql.sources.v2.reader.InputPartition;
-import org.apache.spark.sql.sources.v2.reader.InputPartitionReader;
-
-import java.io.IOException;
-import java.io.UncheckedIOException;
-
-/** A Spark {@link InputPartition} for paimon. */
-public class SparkInputPartition
- implements InputPartition<org.apache.spark.sql.catalyst.InternalRow> {
-
- private static final long serialVersionUID = 1L;
-
- private final ReadBuilder readBuilder;
- private final Split split;
-
- public SparkInputPartition(ReadBuilder readBuilder, Split split) {
- this.readBuilder = readBuilder;
- this.split = split;
- }
-
- @Override
- public InputPartitionReader<org.apache.spark.sql.catalyst.InternalRow> createPartitionReader() {
- RecordReader<InternalRow> recordReader;
- try {
- recordReader = readBuilder.newRead().createReader(split);
- } catch (IOException e) {
- throw new UncheckedIOException(e);
- }
- RecordReaderIterator<InternalRow> iterator = new RecordReaderIterator<>(recordReader);
- SparkInternalRow row = new SparkInternalRow(readBuilder.readType());
- return new SparkInputPartitionReader(iterator, row);
- }
-
- @Override
- public String[] preferredLocations() {
- return new String[0];
- }
-}
diff --git a/paimon-spark/paimon-spark-2/src/main/java/org/apache/paimon/spark/SparkInputPartitionReader.java b/paimon-spark/paimon-spark-2/src/main/java/org/apache/paimon/spark/SparkInputPartitionReader.java
deleted file mode 100644
index d9f925431..000000000
--- a/paimon-spark/paimon-spark-2/src/main/java/org/apache/paimon/spark/SparkInputPartitionReader.java
+++ /dev/null
@@ -1,65 +0,0 @@
-/*
- * 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.paimon.spark;
-
-import org.apache.paimon.reader.RecordReaderIterator;
-
-import org.apache.spark.sql.catalyst.InternalRow;
-import org.apache.spark.sql.sources.v2.reader.InputPartition;
-import org.apache.spark.sql.sources.v2.reader.InputPartitionReader;
-
-import java.io.IOException;
-
-/** A spark 2 {@link InputPartitionReader} for paimon, created by {@link InputPartition}. */
-public class SparkInputPartitionReader implements InputPartitionReader<InternalRow> {
-
- private final RecordReaderIterator<org.apache.paimon.data.InternalRow> iterator;
-
- private final SparkInternalRow row;
-
- public SparkInputPartitionReader(
- RecordReaderIterator<org.apache.paimon.data.InternalRow> iterator,
- SparkInternalRow row) {
- this.iterator = iterator;
- this.row = row;
- }
-
- @Override
- public boolean next() {
- if (iterator.hasNext()) {
- row.replace(iterator.next());
- return true;
- }
- return false;
- }
-
- @Override
- public InternalRow get() {
- return row;
- }
-
- @Override
- public void close() throws IOException {
- try {
- iterator.close();
- } catch (Exception e) {
- throw new IOException(e);
- }
- }
-}
diff --git a/paimon-spark/paimon-spark-2/src/main/java/org/apache/paimon/spark/SparkSource.java b/paimon-spark/paimon-spark-2/src/main/java/org/apache/paimon/spark/SparkSource.java
deleted file mode 100644
index f9e075464..000000000
--- a/paimon-spark/paimon-spark-2/src/main/java/org/apache/paimon/spark/SparkSource.java
+++ /dev/null
@@ -1,62 +0,0 @@
-/*
- * 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.paimon.spark;
-
-import org.apache.paimon.catalog.CatalogContext;
-import org.apache.paimon.options.Options;
-import org.apache.paimon.table.FileStoreTableFactory;
-
-import org.apache.spark.sql.SparkSession;
-import org.apache.spark.sql.sources.DataSourceRegister;
-import org.apache.spark.sql.sources.v2.DataSourceOptions;
-import org.apache.spark.sql.sources.v2.ReadSupport;
-import org.apache.spark.sql.sources.v2.SessionConfigSupport;
-import org.apache.spark.sql.sources.v2.reader.DataSourceReader;
-import org.apache.spark.sql.types.StructType;
-
-/** The Spark source for paimon. */
-public class SparkSource implements DataSourceRegister, ReadSupport, SessionConfigSupport {
-
- /** Not use 'paimon' here, the '-' is not allowed in SQL. */
- private static final String SHORT_NAME = "paimon";
-
- @Override
- public String shortName() {
- return SHORT_NAME;
- }
-
- @Override
- public DataSourceReader createReader(StructType schema, DataSourceOptions options) {
- return createReader(options);
- }
-
- @Override
- public DataSourceReader createReader(DataSourceOptions options) {
- CatalogContext catalogContext =
- CatalogContext.create(
- Options.fromMap(options.asMap()),
- SparkSession.active().sessionState().newHadoopConf());
- return new SparkDataSourceReader(FileStoreTableFactory.create(catalogContext));
- }
-
- @Override
- public String keyPrefix() {
- return SHORT_NAME;
- }
-}
diff --git a/paimon-spark/paimon-spark-2/src/main/resources/META-INF.services/org.apache.spark.sql.sources.DataSourceRegister b/paimon-spark/paimon-spark-2/src/main/resources/META-INF.services/org.apache.spark.sql.sources.DataSourceRegister
deleted file mode 100644
index f2046daf3..000000000
--- a/paimon-spark/paimon-spark-2/src/main/resources/META-INF.services/org.apache.spark.sql.sources.DataSourceRegister
+++ /dev/null
@@ -1,16 +0,0 @@
-# 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.paimon.spark.SparkSource
diff --git a/paimon-spark/paimon-spark-2/src/test/java/org/apache/paimon/spark/SimpleTableTestHelper.java b/paimon-spark/paimon-spark-2/src/test/java/org/apache/paimon/spark/SimpleTableTestHelper.java
deleted file mode 100644
index aa84e34af..000000000
--- a/paimon-spark/paimon-spark-2/src/test/java/org/apache/paimon/spark/SimpleTableTestHelper.java
+++ /dev/null
@@ -1,84 +0,0 @@
-/*
- * 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.paimon.spark;
-
-import org.apache.paimon.CoreOptions;
-import org.apache.paimon.data.InternalRow;
-import org.apache.paimon.fs.Path;
-import org.apache.paimon.fs.local.LocalFileIO;
-import org.apache.paimon.options.Options;
-import org.apache.paimon.schema.Schema;
-import org.apache.paimon.schema.SchemaManager;
-import org.apache.paimon.table.FileStoreTable;
-import org.apache.paimon.table.FileStoreTableFactory;
-import org.apache.paimon.table.sink.StreamTableCommit;
-import org.apache.paimon.table.sink.StreamTableWrite;
-import org.apache.paimon.table.sink.StreamWriteBuilder;
-import org.apache.paimon.types.RowType;
-
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Map;
-
-/** A simple table test helper to write and commit. */
-public class SimpleTableTestHelper {
-
- private final StreamTableWrite writer;
- private final StreamTableCommit commit;
-
- private long commitIdentifier;
-
- public SimpleTableTestHelper(Path path, RowType rowType) throws Exception {
- Map<String, String> options = new HashMap<>();
- // orc is shaded, can not find shaded classes in ide
- options.put(CoreOptions.FILE_FORMAT.key(), CoreOptions.FileFormatType.AVRO.toString());
- options.put("bucket", "1");
- new SchemaManager(LocalFileIO.create(), path)
- .createTable(
- new Schema(
- rowType.getFields(),
- Collections.emptyList(),
- Collections.singletonList(rowType.getFields().get(0).name()),
- options,
- ""));
- Options conf = Options.fromMap(options);
- conf.setString("path", path.toString());
- FileStoreTable table = FileStoreTableFactory.create(LocalFileIO.create(), conf);
-
- StreamWriteBuilder streamWriteBuilder = table.newStreamWriteBuilder();
- writer = streamWriteBuilder.newWrite();
- commit = streamWriteBuilder.newCommit();
-
- this.commitIdentifier = 0;
- }
-
- public void write(InternalRow row) throws Exception {
- writer.write(row);
- }
-
- public void commit() throws Exception {
- commit.commit(commitIdentifier, writer.prepareCommit(true, commitIdentifier));
- commitIdentifier++;
- }
-
- public void close() throws Exception {
- writer.close();
- commit.close();
- }
-}
diff --git a/paimon-spark/paimon-spark-2/src/test/java/org/apache/paimon/spark/SparkFilterConverterTest.java b/paimon-spark/paimon-spark-2/src/test/java/org/apache/paimon/spark/SparkFilterConverterTest.java
deleted file mode 100644
index 2f50fb885..000000000
--- a/paimon-spark/paimon-spark-2/src/test/java/org/apache/paimon/spark/SparkFilterConverterTest.java
+++ /dev/null
@@ -1,175 +0,0 @@
-/*
- * 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.paimon.spark;
-
-import org.apache.paimon.data.Timestamp;
-import org.apache.paimon.predicate.Predicate;
-import org.apache.paimon.predicate.PredicateBuilder;
-import org.apache.paimon.types.DataField;
-import org.apache.paimon.types.DateType;
-import org.apache.paimon.types.IntType;
-import org.apache.paimon.types.RowType;
-import org.apache.paimon.types.TimestampType;
-
-import org.apache.spark.sql.sources.EqualTo;
-import org.apache.spark.sql.sources.GreaterThan;
-import org.apache.spark.sql.sources.GreaterThanOrEqual;
-import org.apache.spark.sql.sources.In;
-import org.apache.spark.sql.sources.IsNotNull;
-import org.apache.spark.sql.sources.IsNull;
-import org.apache.spark.sql.sources.LessThan;
-import org.apache.spark.sql.sources.LessThanOrEqual;
-import org.junit.jupiter.api.Test;
-
-import java.sql.Date;
-import java.time.Instant;
-import java.time.LocalDate;
-import java.time.LocalDateTime;
-import java.time.ZoneOffset;
-import java.util.Arrays;
-import java.util.Collections;
-
-import static org.assertj.core.api.Assertions.assertThat;
-
-/** Test for {@link SparkFilterConverter}. */
-public class SparkFilterConverterTest {
-
- @Test
- public void testAll() {
- RowType rowType =
- new RowType(Collections.singletonList(new DataField(0, "id", new IntType())));
- SparkFilterConverter converter = new SparkFilterConverter(rowType);
- PredicateBuilder builder = new PredicateBuilder(rowType);
-
- String field = "id";
- IsNull isNull = IsNull.apply(field);
- Predicate expectedIsNull = builder.isNull(0);
- Predicate actualIsNull = converter.convert(isNull);
- assertThat(actualIsNull).isEqualTo(expectedIsNull);
-
- IsNotNull isNotNull = IsNotNull.apply(field);
- Predicate expectedIsNotNull = builder.isNotNull(0);
- Predicate actualIsNotNull = converter.convert(isNotNull);
- assertThat(actualIsNotNull).isEqualTo(expectedIsNotNull);
-
- LessThan lt = LessThan.apply(field, 1);
- Predicate expectedLt = builder.lessThan(0, 1);
- Predicate actualLt = converter.convert(lt);
- assertThat(actualLt).isEqualTo(expectedLt);
-
- LessThan ltNull = LessThan.apply(field, null);
- Predicate expectedLtNull = builder.lessThan(0, null);
- Predicate actualLtNull = converter.convert(ltNull);
- assertThat(actualLtNull).isEqualTo(expectedLtNull);
-
- LessThanOrEqual ltEq = LessThanOrEqual.apply(field, 1);
- Predicate expectedLtEq = builder.lessOrEqual(0, 1);
- Predicate actualLtEq = converter.convert(ltEq);
- assertThat(actualLtEq).isEqualTo(expectedLtEq);
-
- LessThanOrEqual ltEqNull = LessThanOrEqual.apply(field, null);
- Predicate expectedLtEqNull = builder.lessOrEqual(0, null);
- Predicate actualLtEqNull = converter.convert(ltEqNull);
- assertThat(actualLtEqNull).isEqualTo(expectedLtEqNull);
-
- GreaterThan gt = GreaterThan.apply(field, 1);
- Predicate expectedGt = builder.greaterThan(0, 1);
- Predicate actualGt = converter.convert(gt);
- assertThat(actualGt).isEqualTo(expectedGt);
-
- GreaterThan gtNull = GreaterThan.apply(field, null);
- Predicate expectedGtNull = builder.greaterThan(0, null);
- Predicate actualGtNull = converter.convert(gtNull);
- assertThat(actualGtNull).isEqualTo(expectedGtNull);
-
- GreaterThanOrEqual gtEq = GreaterThanOrEqual.apply(field, 1);
- Predicate expectedGtEq = builder.greaterOrEqual(0, 1);
- Predicate actualGtEq = converter.convert(gtEq);
- assertThat(actualGtEq).isEqualTo(expectedGtEq);
-
- GreaterThanOrEqual gtEqNull = GreaterThanOrEqual.apply(field, null);
- Predicate expectedGtEqNull = builder.greaterOrEqual(0, null);
- Predicate actualGtEqNull = converter.convert(gtEqNull);
- assertThat(actualGtEqNull).isEqualTo(expectedGtEqNull);
-
- EqualTo eq = EqualTo.apply(field, 1);
- Predicate expectedEq = builder.equal(0, 1);
- Predicate actualEq = converter.convert(eq);
- assertThat(actualEq).isEqualTo(expectedEq);
-
- EqualTo eqNull = EqualTo.apply(field, null);
- Predicate expectedEqNull = builder.equal(0, null);
- Predicate actualEqNull = converter.convert(eqNull);
- assertThat(actualEqNull).isEqualTo(expectedEqNull);
-
- In in = In.apply(field, new Object[] {1, null, 2});
- Predicate expectedIn = builder.in(0, Arrays.asList(1, null, 2));
- Predicate actualIn = converter.convert(in);
- assertThat(actualIn).isEqualTo(expectedIn);
-
- Object[] literals = new Object[30];
- literals[0] = null;
- for (int i = 1; i < literals.length; i++) {
- literals[i] = i * 100;
- }
- In largeIn = In.apply(field, literals);
- Predicate expectedLargeIn = builder.in(0, Arrays.asList(literals));
- Predicate actualLargeIn = converter.convert(largeIn);
- assertThat(actualLargeIn).isEqualTo(expectedLargeIn);
- }
-
- @Test
- public void testTimestamp() {
- RowType rowType =
- new RowType(Collections.singletonList(new DataField(0, "x", new TimestampType())));
- SparkFilterConverter converter = new SparkFilterConverter(rowType);
- PredicateBuilder builder = new PredicateBuilder(rowType);
-
- java.sql.Timestamp timestamp = java.sql.Timestamp.valueOf("2018-10-18 00:00:57.907");
- LocalDateTime localDateTime = LocalDateTime.parse("2018-10-18T00:00:57.907");
- Instant instant = localDateTime.toInstant(ZoneOffset.UTC);
-
- Predicate instantExpression = converter.convert(GreaterThan.apply("x", instant));
- Predicate timestampExpression = converter.convert(GreaterThan.apply("x", timestamp));
- Predicate rawExpression =
- builder.greaterThan(0, Timestamp.fromLocalDateTime(localDateTime));
-
- assertThat(timestampExpression).isEqualTo(rawExpression);
- assertThat(instantExpression).isEqualTo(rawExpression);
- }
-
- @Test
- public void testDate() {
- RowType rowType =
- new RowType(Collections.singletonList(new DataField(0, "x", new DateType())));
- SparkFilterConverter converter = new SparkFilterConverter(rowType);
- PredicateBuilder builder = new PredicateBuilder(rowType);
-
- LocalDate localDate = LocalDate.parse("2018-10-18");
- Date date = Date.valueOf(localDate);
- int epochDay = (int) localDate.toEpochDay();
-
- Predicate localDateExpression = converter.convert(GreaterThan.apply("x", localDate));
- Predicate dateExpression = converter.convert(GreaterThan.apply("x", date));
- Predicate rawExpression = builder.greaterThan(0, epochDay);
-
- assertThat(dateExpression).isEqualTo(rawExpression);
- assertThat(localDateExpression).isEqualTo(rawExpression);
- }
-}
diff --git a/paimon-spark/paimon-spark-2/src/test/java/org/apache/paimon/spark/SparkInternalRowTest.java b/paimon-spark/paimon-spark-2/src/test/java/org/apache/paimon/spark/SparkInternalRowTest.java
deleted file mode 100644
index 1c368c6ec..000000000
--- a/paimon-spark/paimon-spark-2/src/test/java/org/apache/paimon/spark/SparkInternalRowTest.java
+++ /dev/null
@@ -1,117 +0,0 @@
-/*
- * 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.paimon.spark;
-
-import org.apache.paimon.data.BinaryString;
-import org.apache.paimon.data.Decimal;
-import org.apache.paimon.data.GenericArray;
-import org.apache.paimon.data.GenericMap;
-import org.apache.paimon.data.GenericRow;
-import org.apache.paimon.data.InternalRow;
-import org.apache.paimon.data.Timestamp;
-import org.apache.paimon.utils.DateTimeUtils;
-
-import org.apache.spark.sql.catalyst.CatalystTypeConverters;
-import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema;
-import org.junit.jupiter.api.Test;
-
-import java.math.BigDecimal;
-import java.nio.charset.StandardCharsets;
-import java.sql.Date;
-import java.time.LocalDate;
-import java.time.LocalDateTime;
-import java.util.AbstractMap;
-import java.util.Arrays;
-import java.util.Map;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
-
-import scala.Function1;
-
-import static org.apache.paimon.data.BinaryString.fromString;
-import static org.apache.paimon.spark.SparkTypeTest.ALL_TYPES;
-import static org.assertj.core.api.Assertions.assertThat;
-
-/** Test for {@link SparkInternalRow}. */
-public class SparkInternalRowTest {
-
- @Test
- public void test() {
- InternalRow rowData =
- GenericRow.of(
- 1,
- fromString("jingsong"),
- 22.2,
- new GenericMap(
- Stream.of(
- new AbstractMap.SimpleEntry<>(
- fromString("key1"),
- GenericRow.of(1.2, 2.3)),
- new AbstractMap.SimpleEntry<>(
- fromString("key2"),
- GenericRow.of(2.4, 3.5)))
- .collect(
- Collectors.toMap(
- Map.Entry::getKey, Map.Entry::getValue))),
- new GenericArray(new BinaryString[] {fromString("v1"), fromString("v5")}),
- new GenericArray(new Integer[] {10, 30}),
- true,
- (byte) 22,
- (short) 356,
- 23567222L,
- "varbinary_v".getBytes(StandardCharsets.UTF_8),
- Timestamp.fromLocalDateTime(LocalDateTime.parse("2007-12-03T10:15:30")),
- DateTimeUtils.toInternal(LocalDate.parse("2022-05-02")),
- Decimal.fromBigDecimal(BigDecimal.valueOf(0.21), 2, 2),
- Decimal.fromBigDecimal(BigDecimal.valueOf(65782123123.01), 38, 2),
- Decimal.fromBigDecimal(BigDecimal.valueOf(62123123.5), 10, 1));
-
- Function1<Object, Object> sparkConverter =
- CatalystTypeConverters.createToScalaConverter(
- SparkTypeUtils.fromPaimonType(ALL_TYPES));
- org.apache.spark.sql.Row sparkRow =
- (org.apache.spark.sql.Row)
- sparkConverter.apply(new SparkInternalRow(ALL_TYPES).replace(rowData));
-
- assertThat(sparkRow.getInt(0)).isEqualTo(1);
- assertThat(sparkRow.getString(1)).isEqualTo("jingsong");
- assertThat(sparkRow.getDouble(2)).isEqualTo(22.2);
- assertThat(((GenericRowWithSchema) sparkRow.getJavaMap(3).get("key1")).values())
- .isEqualTo(new Object[] {1.2, 2.3});
- assertThat(((GenericRowWithSchema) sparkRow.getJavaMap(3).get("key2")).values())
- .isEqualTo(new Object[] {2.4, 3.5});
- assertThat(sparkRow.getList(4)).isEqualTo(Arrays.asList("v1", "v5"));
- assertThat(sparkRow.getList(5)).isEqualTo(Arrays.asList(10, 30));
- assertThat(sparkRow.getBoolean(6)).isEqualTo(true);
- assertThat(sparkRow.getByte(7)).isEqualTo((byte) 22);
- assertThat(sparkRow.getShort(8)).isEqualTo((short) 356);
- assertThat(sparkRow.getLong(9)).isEqualTo(23567222L);
- assertThat((byte[]) sparkRow.get(10))
- .isEqualTo("varbinary_v".getBytes(StandardCharsets.UTF_8));
- assertThat(sparkRow.getTimestamp(11).getTime())
- .isEqualTo(
- java.sql.Timestamp.valueOf(LocalDateTime.parse("2007-12-03T10:15:30"))
- .getTime());
- assertThat(sparkRow.getDate(12).getTime())
- .isEqualTo(Date.valueOf(LocalDate.parse("2022-05-02")).getTime());
- assertThat(sparkRow.getDecimal(13)).isEqualTo(BigDecimal.valueOf(0.21));
- assertThat(sparkRow.getDecimal(14)).isEqualTo(BigDecimal.valueOf(65782123123.01));
- assertThat(sparkRow.getDecimal(15)).isEqualTo(BigDecimal.valueOf(62123123.5));
- }
-}
diff --git a/paimon-spark/paimon-spark-2/src/test/java/org/apache/paimon/spark/SparkReadITCase.java b/paimon-spark/paimon-spark-2/src/test/java/org/apache/paimon/spark/SparkReadITCase.java
deleted file mode 100644
index 0f7d33339..000000000
--- a/paimon-spark/paimon-spark-2/src/test/java/org/apache/paimon/spark/SparkReadITCase.java
+++ /dev/null
@@ -1,133 +0,0 @@
-/*
- * 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.paimon.spark;
-
-import org.apache.paimon.data.BinaryString;
-import org.apache.paimon.data.GenericRow;
-import org.apache.paimon.fs.Path;
-import org.apache.paimon.types.BigIntType;
-import org.apache.paimon.types.DataField;
-import org.apache.paimon.types.IntType;
-import org.apache.paimon.types.RowKind;
-import org.apache.paimon.types.RowType;
-import org.apache.paimon.types.VarCharType;
-
-import org.apache.commons.io.FileUtils;
-import org.apache.spark.sql.Dataset;
-import org.apache.spark.sql.Row;
-import org.apache.spark.sql.SparkSession;
-import org.junit.jupiter.api.AfterAll;
-import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.Test;
-
-import java.io.File;
-import java.io.IOException;
-import java.nio.file.Files;
-import java.util.Arrays;
-import java.util.List;
-
-import static org.assertj.core.api.Assertions.assertThat;
-
-/** ITCase for Spark reader. */
-public class SparkReadITCase {
-
- private static File warehouse = null;
-
- private static SparkSession spark = null;
-
- private static Path tablePath1;
-
- private static Path tablePath2;
-
- @BeforeAll
- public static void startSpark() throws Exception {
- warehouse = Files.createTempFile("warehouse", null).toFile();
- assertThat(warehouse.delete()).isTrue();
- Path warehousePath = new Path("file:" + warehouse);
- spark = SparkSession.builder().master("local[2]").getOrCreate();
-
- // Paimon sink
- tablePath1 = new Path(warehousePath, "default.db/t1");
- SimpleTableTestHelper testHelper1 = createTestHelper(tablePath1);
- testHelper1.write(GenericRow.of(1, 2L, BinaryString.fromString("1")));
- testHelper1.write(GenericRow.of(3, 4L, BinaryString.fromString("2")));
- testHelper1.write(GenericRow.of(5, 6L, BinaryString.fromString("3")));
- testHelper1.write(GenericRow.ofKind(RowKind.DELETE, 3, 4L, BinaryString.fromString("2")));
- testHelper1.commit();
- testHelper1.close();
-
- tablePath2 = new Path(warehousePath, "default.db/t2");
- SimpleTableTestHelper testHelper2 = createTestHelper(tablePath2);
- testHelper2.write(GenericRow.of(1, 2L, BinaryString.fromString("1")));
- testHelper2.write(GenericRow.of(3, 4L, BinaryString.fromString("2")));
- testHelper2.commit();
- testHelper2.write(GenericRow.of(5, 6L, BinaryString.fromString("3")));
- testHelper2.write(GenericRow.of(7, 8L, BinaryString.fromString("4")));
- testHelper2.commit();
- testHelper2.close();
- }
-
- private static SimpleTableTestHelper createTestHelper(Path tablePath) throws Exception {
- RowType rowType =
- new RowType(
- Arrays.asList(
- new DataField(0, "a", new IntType()),
- new DataField(1, "b", new BigIntType()),
- new DataField(2, "c", new VarCharType())));
- return new SimpleTableTestHelper(tablePath, rowType);
- }
-
- @AfterAll
- public static void stopSpark() throws IOException {
- if (warehouse != null && warehouse.exists()) {
- FileUtils.deleteDirectory(warehouse);
- }
- if (spark != null) {
- spark.stop();
- spark = null;
- }
- }
-
- @Test
- public void testNormal() {
- innerTestNormal(spark.read().format("paimon").option("path", tablePath1.toString()).load());
- }
-
- @Test
- public void testFilterPushDown() {
- innerTestFilterPushDown(
- spark.read().format("paimon").option("path", tablePath2.toString()).load());
- }
-
- private void innerTestNormal(Dataset<Row> dataset) {
- List<Row> results = dataset.collectAsList();
- assertThat(results.toString()).isEqualTo("[[1,2,1], [5,6,3]]");
-
- results = dataset.select("a", "c").collectAsList();
- assertThat(results.toString()).isEqualTo("[[1,1], [5,3]]");
-
- results = dataset.groupBy().sum("b").collectAsList();
- assertThat(results.toString()).isEqualTo("[[8]]");
- }
-
- private void innerTestFilterPushDown(Dataset<Row> dataset) {
- List<Row> results = dataset.filter("a < 4").select("a", "c").collectAsList();
- assertThat(results.toString()).isEqualTo("[[1,1], [3,2]]");
- }
-}
diff --git a/paimon-spark/paimon-spark-2/src/test/java/org/apache/paimon/spark/SparkTypeTest.java b/paimon-spark/paimon-spark-2/src/test/java/org/apache/paimon/spark/SparkTypeTest.java
deleted file mode 100644
index a99a6ff37..000000000
--- a/paimon-spark/paimon-spark-2/src/test/java/org/apache/paimon/spark/SparkTypeTest.java
+++ /dev/null
@@ -1,96 +0,0 @@
-/*
- * 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.paimon.spark;
-
-import org.apache.paimon.types.DataTypes;
-import org.apache.paimon.types.RowType;
-
-import org.apache.spark.sql.types.StructType;
-import org.junit.jupiter.api.Test;
-
-import static org.apache.paimon.spark.SparkTypeUtils.fromPaimonRowType;
-import static org.assertj.core.api.Assertions.assertThat;
-
-/** Test for {@link SparkTypeUtils}. */
-public class SparkTypeTest {
-
- public static final RowType ALL_TYPES =
- RowType.builder()
- .field("id", DataTypes.INT().notNull())
- .field("name", DataTypes.STRING()) /* optional by default */
- .field("salary", DataTypes.DOUBLE().notNull())
- .field(
- "locations",
- DataTypes.MAP(
- DataTypes.STRING().notNull(),
- DataTypes.ROW(
- DataTypes.FIELD(
- 0,
- "posX",
- DataTypes.DOUBLE().notNull(),
- "X field"),
- DataTypes.FIELD(
- 1,
- "posY",
- DataTypes.DOUBLE().notNull(),
- "Y field"))))
- .field("strArray", DataTypes.ARRAY(DataTypes.STRING()).nullable())
- .field("intArray", DataTypes.ARRAY(DataTypes.INT()).nullable())
- .field("boolean", DataTypes.BOOLEAN().nullable())
- .field("tinyint", DataTypes.TINYINT())
- .field("smallint", DataTypes.SMALLINT())
- .field("bigint", DataTypes.BIGINT())
- .field("bytes", DataTypes.BYTES())
- .field("timestamp", DataTypes.TIMESTAMP())
- .field("date", DataTypes.DATE())
- .field("decimal", DataTypes.DECIMAL(2, 2))
- .field("decimal2", DataTypes.DECIMAL(38, 2))
- .field("decimal3", DataTypes.DECIMAL(10, 1))
- .build();
-
- @Test
- public void testAllTypes() {
- String nestedRowMapType =
- "StructField(locations,MapType("
- + "StringType,"
- + "StructType(StructField(posX,DoubleType,true), StructField(posY,DoubleType,true)),true),true)";
- String expected =
- "StructType("
- + "StructField(id,IntegerType,true), "
- + "StructField(name,StringType,true), "
- + "StructField(salary,DoubleType,true), "
- + nestedRowMapType
- + ", "
- + "StructField(strArray,ArrayType(StringType,true),true), "
- + "StructField(intArray,ArrayType(IntegerType,true),true), "
- + "StructField(boolean,BooleanType,true), "
- + "StructField(tinyint,ByteType,true), "
- + "StructField(smallint,ShortType,true), "
- + "StructField(bigint,LongType,true), "
- + "StructField(bytes,BinaryType,true), "
- + "StructField(timestamp,TimestampType,true), "
- + "StructField(date,DateType,true), "
- + "StructField(decimal,DecimalType(2,2),true), "
- + "StructField(decimal2,DecimalType(38,2),true), "
- + "StructField(decimal3,DecimalType(10,1),true))";
-
- StructType sparkType = fromPaimonRowType(ALL_TYPES);
- assertThat(sparkType.toString()).isEqualTo(expected);
- }
-}
diff --git a/paimon-spark/paimon-spark-2/src/test/resources/META-INF/services/org.junit.jupiter.api.extension.Extension b/paimon-spark/paimon-spark-2/src/test/resources/META-INF/services/org.junit.jupiter.api.extension.Extension
deleted file mode 100644
index c9cb1908a..000000000
--- a/paimon-spark/paimon-spark-2/src/test/resources/META-INF/services/org.junit.jupiter.api.extension.Extension
+++ /dev/null
@@ -1,16 +0,0 @@
-# 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.paimon.testutils.junit.TestLoggerExtension
\ No newline at end of file
diff --git a/paimon-spark/paimon-spark-2/src/test/resources/log4j2-test.properties b/paimon-spark/paimon-spark-2/src/test/resources/log4j2-test.properties
deleted file mode 100644
index 6f324f586..000000000
--- a/paimon-spark/paimon-spark-2/src/test/resources/log4j2-test.properties
+++ /dev/null
@@ -1,38 +0,0 @@
-################################################################################
-# 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 [%tid %t] %-5p %c %x - %m%n
-
-logger.kafka.name = kafka
-logger.kafka.level = OFF
-logger.kafka2.name = state.change
-logger.kafka2.level = OFF
-
-logger.zookeeper.name = org.apache.zookeeper
-logger.zookeeper.level = OFF
-logger.I0Itec.name = org.I0Itec
-logger.I0Itec.level = OFF
diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkInternalRow.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkInternalRow.java
index a9306e984..a54f3e339 100644
--- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkInternalRow.java
+++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkInternalRow.java
@@ -36,7 +36,23 @@ import org.apache.spark.sql.catalyst.util.ArrayBasedMapData;
import org.apache.spark.sql.catalyst.util.ArrayData;
import org.apache.spark.sql.catalyst.util.DateTimeUtils;
import org.apache.spark.sql.catalyst.util.MapData;
+import org.apache.spark.sql.types.BinaryType;
+import org.apache.spark.sql.types.BooleanType;
+import org.apache.spark.sql.types.ByteType;
+import org.apache.spark.sql.types.CalendarIntervalType;
+import org.apache.spark.sql.types.DateType;
import org.apache.spark.sql.types.Decimal;
+import org.apache.spark.sql.types.DecimalType;
+import org.apache.spark.sql.types.DoubleType;
+import org.apache.spark.sql.types.FloatType;
+import org.apache.spark.sql.types.IntegerType;
+import org.apache.spark.sql.types.LongType;
+import org.apache.spark.sql.types.NullType;
+import org.apache.spark.sql.types.ShortType;
+import org.apache.spark.sql.types.StringType;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.sql.types.TimestampType;
+import org.apache.spark.sql.types.UserDefinedType;
import org.apache.spark.unsafe.types.CalendarInterval;
import org.apache.spark.unsafe.types.UTF8String;
@@ -165,7 +181,63 @@ public class SparkInternalRow extends org.apache.spark.sql.catalyst.InternalRow
@Override
public Object get(int ordinal, org.apache.spark.sql.types.DataType dataType) {
- return SpecializedGettersReader.read(this, ordinal, dataType);
+ if (isNullAt(ordinal) || dataType instanceof NullType) {
+ return null;
+ }
+ if (dataType instanceof BooleanType) {
+ return getBoolean(ordinal);
+ }
+ if (dataType instanceof ByteType) {
+ return getByte(ordinal);
+ }
+ if (dataType instanceof ShortType) {
+ return getShort(ordinal);
+ }
+ if (dataType instanceof IntegerType) {
+ return getInt(ordinal);
+ }
+ if (dataType instanceof LongType) {
+ return getLong(ordinal);
+ }
+ if (dataType instanceof FloatType) {
+ return getFloat(ordinal);
+ }
+ if (dataType instanceof DoubleType) {
+ return getDouble(ordinal);
+ }
+ if (dataType instanceof StringType) {
+ return getUTF8String(ordinal);
+ }
+ if (dataType instanceof DecimalType) {
+ DecimalType dt = (DecimalType) dataType;
+ return getDecimal(ordinal, dt.precision(), dt.scale());
+ }
+ if (dataType instanceof DateType) {
+ return getInt(ordinal);
+ }
+ if (dataType instanceof TimestampType) {
+ return getLong(ordinal);
+ }
+ if (dataType instanceof CalendarIntervalType) {
+ return getInterval(ordinal);
+ }
+ if (dataType instanceof BinaryType) {
+ return getBinary(ordinal);
+ }
+ if (dataType instanceof StructType) {
+ return getStruct(ordinal, ((StructType) dataType).size());
+ }
+ if (dataType instanceof org.apache.spark.sql.types.ArrayType) {
+ return getArray(ordinal);
+ }
+ if (dataType instanceof org.apache.spark.sql.types.MapType) {
+ return getMap(ordinal);
+ }
+ if (dataType instanceof UserDefinedType) {
+ return get(ordinal, ((UserDefinedType<?>) dataType).sqlType());
+ }
+
+ throw new UnsupportedOperationException("Unsupported data type " + dataType.simpleString());
}
public static Object fromPaimon(Object o, DataType type) {
diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SpecializedGettersReader.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SpecializedGettersReader.java
deleted file mode 100644
index 388d10bfd..000000000
--- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SpecializedGettersReader.java
+++ /dev/null
@@ -1,109 +0,0 @@
-/*
- * 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.paimon.spark;
-
-import org.apache.spark.sql.catalyst.expressions.SpecializedGetters;
-import org.apache.spark.sql.types.ArrayType;
-import org.apache.spark.sql.types.BinaryType;
-import org.apache.spark.sql.types.BooleanType;
-import org.apache.spark.sql.types.ByteType;
-import org.apache.spark.sql.types.CalendarIntervalType;
-import org.apache.spark.sql.types.DataType;
-import org.apache.spark.sql.types.DateType;
-import org.apache.spark.sql.types.DecimalType;
-import org.apache.spark.sql.types.DoubleType;
-import org.apache.spark.sql.types.FloatType;
-import org.apache.spark.sql.types.IntegerType;
-import org.apache.spark.sql.types.LongType;
-import org.apache.spark.sql.types.MapType;
-import org.apache.spark.sql.types.NullType;
-import org.apache.spark.sql.types.ShortType;
-import org.apache.spark.sql.types.StringType;
-import org.apache.spark.sql.types.StructType;
-import org.apache.spark.sql.types.TimestampType;
-import org.apache.spark.sql.types.UserDefinedType;
-
-/**
- * Reader of Spark {@link SpecializedGetters}. Copied from Spark to avoid conflict between Spark2
- * and Spark3 .
- */
-public final class SpecializedGettersReader {
-
- private SpecializedGettersReader() {}
-
- public static Object read(SpecializedGetters obj, int ordinal, DataType dataType) {
- if (obj.isNullAt(ordinal) || dataType instanceof NullType) {
- return null;
- }
- if (dataType instanceof BooleanType) {
- return obj.getBoolean(ordinal);
- }
- if (dataType instanceof ByteType) {
- return obj.getByte(ordinal);
- }
- if (dataType instanceof ShortType) {
- return obj.getShort(ordinal);
- }
- if (dataType instanceof IntegerType) {
- return obj.getInt(ordinal);
- }
- if (dataType instanceof LongType) {
- return obj.getLong(ordinal);
- }
- if (dataType instanceof FloatType) {
- return obj.getFloat(ordinal);
- }
- if (dataType instanceof DoubleType) {
- return obj.getDouble(ordinal);
- }
- if (dataType instanceof StringType) {
- return obj.getUTF8String(ordinal);
- }
- if (dataType instanceof DecimalType) {
- DecimalType dt = (DecimalType) dataType;
- return obj.getDecimal(ordinal, dt.precision(), dt.scale());
- }
- if (dataType instanceof DateType) {
- return obj.getInt(ordinal);
- }
- if (dataType instanceof TimestampType) {
- return obj.getLong(ordinal);
- }
- if (dataType instanceof CalendarIntervalType) {
- return obj.getInterval(ordinal);
- }
- if (dataType instanceof BinaryType) {
- return obj.getBinary(ordinal);
- }
- if (dataType instanceof StructType) {
- return obj.getStruct(ordinal, ((StructType) dataType).size());
- }
- if (dataType instanceof ArrayType) {
- return obj.getArray(ordinal);
- }
- if (dataType instanceof MapType) {
- return obj.getMap(ordinal);
- }
- if (dataType instanceof UserDefinedType) {
- return obj.get(ordinal, ((UserDefinedType) dataType).sqlType());
- }
-
- throw new UnsupportedOperationException("Unsupported data type " + dataType.simpleString());
- }
-}
diff --git a/paimon-spark/pom.xml b/paimon-spark/pom.xml
index fe2aa5024..6875a8bf3 100644
--- a/paimon-spark/pom.xml
+++ b/paimon-spark/pom.xml
@@ -39,7 +39,6 @@ under the License.
<modules>
<module>paimon-spark-common</module>
- <module>paimon-spark-2</module>
<module>paimon-spark-3.5</module>
<module>paimon-spark-3.4</module>
<module>paimon-spark-3.3</module>