You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2024/05/30 07:39:52 UTC
(paimon) branch release-0.8 updated (276d1ab51 -> cfcc72d7d)
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a change to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/paimon.git
from 276d1ab51 [flink] Introduce ManifestCommittableSerializerCompatibilityTest
new 499e9c5db [minor] Add Nullable to logOffsets in Snapshot
new ad371d5ca [minor] Fix exception message in FileStoreCommitImpl
new 64e031343 [doc] add some descriptions about data writing in basic-concepts (#3301)
new 580208828 [cdc] Fixed when the order of the same field differs, it is considered a schema change. (#3314)
new d83a06255 [doc] Reorg Append table pages
new 45be92ac2 [hive] fix db-location conflict with location-in-properties (#3309)
new a177de33a [doc] Update version and link of Doris and Trino (#3322)
new 3c75ca773 [spark] Fix drop database cascade when database is not empy with SparkGenericCatalog (#3324)
new 92e2f00f8 [core] FileStoreWrite.State should get max sequence number from RecordWriter instead of from data files (#3327)
new 508b260fd [spark] Add check for provider when creating table with SparkCatalog (#3326)
new 636b9fa58 [cdc] Add more checker for the timestamp type. (#3312)
new b8b6e8664 [doc] Optimize useless message in dedicated-compaction
new 51103e043 [core] Add JsonIgnoreProperties to all JSON class
new a088a044c [doc] Add Specification Page
new 688d12adf [core] Avoid potential NPE in FileStoreCommitImpl and TagAutoCreation
new ee3cd40a1 [bundle] add source code to the source jar for paimon-bundle (#3333)
new b1944274e [flink] Respect waitCompaction in FlinkCdcMultiTableSink (#3337)
new de32dae41 [core] Incremental-between tags should deduplicate records (#3338)
new 9e63ea6e2 [spark][core] fix when a table have no snapshot or no statistics (#3341)
new 0b63cb8a9 [spark] Fix drop database cascade when database with view with SparkGenericCatalog (#3347)
new 3881d6a88 [core] Revert branchs in TagsTable (#3350)
new e3e4303ab [core] Fix that ignore-delete hasn't handle rowkind.field (#3365)
new 78a850937 [flink] Adjust the setMaxParallelism api call in QueryService (#3377)
new f91a22029 [spark] Fix HiveMigrator hive api version compatible issue (#3381)
new 6355d13a0 [flink] Cleanup QueryService file after job is down (#3382)
new 638e89a4a [doc] Add doc for StarRocks (#3392)
new 67db033e8 [doc] Remove outdate roadmap page
new 42299afdc [doc] Fix typo in mongo-cdc.md (#3413)
new f3e67b7b3 [doc] Update managed memory guide for write buffer managing (#3417)
new d78658a18 [spark] SparkHilbertUDF hilbertCurvePosBytes classnotfound fix (#3415)
new cfcc72d7d [hive] Fix flink not synchronizing to hive metastore when deleting table partitions using hive catalog (#3411)
The 31 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../{append-queue-table.md => append-queue.md} | 48 ++--
docs/content/append-table/append-scalable-table.md | 114 ----------
docs/content/append-table/append-table.md | 64 ++++++
docs/content/concepts/basic-concepts.md | 2 +-
docs/content/concepts/specification.md | 253 +++++++++++++++++++++
docs/content/engines/overview.md | 9 +-
docs/content/engines/starrocks.md | 186 +++++++++++++++
docs/content/flink/cdc-ingestion/mongo-cdc.md | 4 +-
docs/content/learn-paimon/understand-files.md | 2 +-
docs/content/maintenance/dedicated-compaction.md | 10 +-
docs/content/maintenance/system-tables.md | 2 +-
docs/content/maintenance/write-performance.md | 4 +
docs/content/migration/migration-from-hive.md | 2 +-
docs/content/project/roadmap.md | 48 ----
docs/static/img/for-scalable.png | Bin 1331757 -> 0 bytes
paimon-bundle/pom.xml | 66 ++++++
.../src/assembly/src-assembly.xml | 37 +--
.../apache/paimon/sort/hilbert/HilbertIndexer.java | 2 +-
.../src/main/java/org/apache/paimon/Changelog.java | 2 +
.../src/main/java/org/apache/paimon/Snapshot.java | 3 +-
.../org/apache/paimon/append/AppendOnlyWriter.java | 5 +
.../java/org/apache/paimon/consumer/Consumer.java | 2 +
.../org/apache/paimon/mergetree/MergeSorter.java | 4 +
.../apache/paimon/mergetree/MergeTreeWriter.java | 5 +
.../paimon/operation/AbstractFileStoreWrite.java | 6 +
.../paimon/operation/AppendOnlyFileStoreWrite.java | 6 +-
.../paimon/operation/FileStoreCommitImpl.java | 7 +-
.../apache/paimon/operation/FileStoreWrite.java | 6 +-
.../paimon/operation/KeyValueFileStoreWrite.java | 4 +-
.../paimon/operation/MergeFileSplitRead.java | 109 ++++-----
.../apache/paimon/operation/RawFileSplitRead.java | 12 +
.../org/apache/paimon/operation/SplitRead.java | 46 +++-
.../java/org/apache/paimon/stats/ColStats.java | 2 +
.../java/org/apache/paimon/stats/Statistics.java | 2 +
.../paimon/table/AbstractFileStoreTable.java | 5 +
.../paimon/table/AppendOnlyFileStoreTable.java | 7 +-
.../paimon/table/PrimaryKeyFileStoreTable.java | 21 +-
.../apache/paimon/table/PrimaryKeyTableUtils.java | 8 +-
.../apache/paimon/table/sink/RowKindGenerator.java | 4 +
.../apache/paimon/table/sink/TableWriteImpl.java | 16 +-
.../paimon/table/source/KeyValueTableRead.java | 79 +++----
.../IncrementalChangelogReadProvider.java | 96 ++++++++
.../splitread/IncrementalDiffReadProvider.java | 65 ++++++
.../splitread/IncrementalDiffSplitRead.java} | 95 +++++++-
.../splitread/MergeFileSplitReadProvider.java | 68 ++++++
.../source/splitread/RawFileSplitReadProvider.java | 60 +++++
.../table/source/splitread/SplitReadProvider.java | 14 +-
.../apache/paimon/table/system/StatisticTable.java | 26 ++-
.../org/apache/paimon/table/system/TagsTable.java | 35 +--
.../src/main/java/org/apache/paimon/tag/Tag.java | 2 +
.../org/apache/paimon/tag/TagAutoCreation.java | 8 +-
.../java/org/apache/paimon/utils/RecordWriter.java | 3 +
.../paimon/operation/MergeFileSplitReadTest.java | 2 +-
.../apache/paimon/table/IncrementalTableTest.java | 18 +-
.../apache/paimon/table/system/TagsTableTest.java | 35 +--
.../cdc/format/debezium/CdcDebeziumSchema.java | 91 --------
.../action/cdc/mysql/format/DebeziumEvent.java | 2 +
.../flink/sink/cdc/FlinkCdcMultiTableSink.java | 4 +-
.../paimon/flink/sink/cdc/RichEventParser.java | 17 +-
.../cdc/UpdatedDataFieldsProcessFunctionBase.java | 11 +
.../cdc/mysql/MySqlSyncTableActionITCase.java | 42 ++++
.../UpdatedDataFieldsProcessFunctionBaseTest.java | 19 ++
.../src/test/resources/mysql/sync_table_setup.sql | 11 +
.../paimon/flink/service/QueryAddressRegister.java | 14 +-
.../paimon/flink/service/QueryFileMonitor.java | 5 +
.../apache/paimon/flink/service/QueryService.java | 19 +-
.../paimon/flink/sink/LocalMergeOperator.java | 3 +-
.../apache/paimon/flink/BatchFileStoreITCase.java | 16 ++
.../paimon/flink/RemoteLookupJoinITCase.java | 31 +++
.../java/org/apache/paimon/hive/HiveCatalog.java | 20 ++
.../paimon/hive/TBPropertiesLocationHelper.java | 1 +
.../apache/paimon/hive/migrate/HiveMigrator.java | 9 +-
.../apache/paimon/hive/HiveCatalogITCaseBase.java | 52 +++++
paimon-spark/paimon-spark-common/pom.xml | 5 -
.../java/org/apache/paimon/spark/SparkCatalog.java | 17 +-
.../apache/paimon/spark/SparkGenericCatalog.java | 16 +-
.../paimon/spark/catalog/SparkBaseCatalog.java | 5 +
.../apache/paimon/spark/sort/SparkHilbertUDF.java | 14 +-
.../commands/PaimonAnalyzeTableColumnCommand.scala | 11 +-
.../apache/paimon/spark/PaimonHiveTestBase.scala | 10 +-
.../apache/paimon/spark/PaimonSparkTestBase.scala | 13 +-
.../paimon/spark/sql/AnalyzeTableTestBase.scala | 8 +
.../org/apache/paimon/spark/sql/DDLTestBase.scala | 13 ++
.../spark/sql/DDLWithHiveCatalogTestBase.scala | 31 ++-
.../paimon/spark/sql/SparkVersionSupport.scala} | 7 +-
85 files changed, 1646 insertions(+), 612 deletions(-)
rename docs/content/append-table/{append-queue-table.md => append-queue.md} (92%)
delete mode 100644 docs/content/append-table/append-scalable-table.md
create mode 100644 docs/content/append-table/append-table.md
create mode 100644 docs/content/concepts/specification.md
create mode 100644 docs/content/engines/starrocks.md
delete mode 100644 docs/content/project/roadmap.md
delete mode 100644 docs/static/img/for-scalable.png
rename docs/content/append-table/overview.md => paimon-bundle/src/assembly/src-assembly.xml (51%)
create mode 100644 paimon-core/src/main/java/org/apache/paimon/table/source/splitread/IncrementalChangelogReadProvider.java
create mode 100644 paimon-core/src/main/java/org/apache/paimon/table/source/splitread/IncrementalDiffReadProvider.java
rename paimon-core/src/main/java/org/apache/paimon/{operation/DiffReader.java => table/source/splitread/IncrementalDiffSplitRead.java} (55%)
create mode 100644 paimon-core/src/main/java/org/apache/paimon/table/source/splitread/MergeFileSplitReadProvider.java
create mode 100644 paimon-core/src/main/java/org/apache/paimon/table/source/splitread/RawFileSplitReadProvider.java
copy paimon-common/src/main/java/org/apache/paimon/utils/FieldsComparator.java => paimon-core/src/main/java/org/apache/paimon/table/source/splitread/SplitReadProvider.java (70%)
delete mode 100644 paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/debezium/CdcDebeziumSchema.java
copy paimon-spark/{paimon-spark-3.3/src/test/scala/org/apache/paimon/spark/sql/AnalyzeTableTest.scala => paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/SparkVersionSupport.scala} (83%)
(paimon) 26/31: [doc] Add doc for StarRocks (#3392)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 638e89a4ad48a0470e30ebe70ae849ea59378814
Author: Jiao Mingye <35...@users.noreply.github.com>
AuthorDate: Fri May 24 18:14:43 2024 +0800
[doc] Add doc for StarRocks (#3392)
---
docs/content/engines/starrocks.md | 186 ++++++++++++++++++++++++++++++
docs/content/maintenance/system-tables.md | 2 +-
2 files changed, 187 insertions(+), 1 deletion(-)
diff --git a/docs/content/engines/starrocks.md b/docs/content/engines/starrocks.md
new file mode 100644
index 000000000..d646dc5d9
--- /dev/null
+++ b/docs/content/engines/starrocks.md
@@ -0,0 +1,186 @@
+---
+title: "StarRocks"
+weight: 5
+type: docs
+aliases:
+- /engines/starrocks.html
+---
+<!--
+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.
+-->
+
+# StarRocks
+
+This documentation is a guide for using Paimon in StarRocks.
+
+## Version
+
+Paimon currently supports StarRocks 3.1 and above. Recommended version is StarRocks 3.2.6 or above.
+
+## Create Paimon Catalog
+
+Paimon catalogs are registered by executing a `CREATE EXTERNAL CATALOG` SQL in StarRocks.
+For example, you can use the following SQL to create a Paimon catalog named paimon_catalog.
+
+```sql
+CREATE EXTERNAL CATALOG paimon_catalog
+PROPERTIES
+(
+"type" = "paimon",
+"paimon.catalog.type" = "filesystem",
+"paimon.catalog.warehouse" = "oss://<your_bucket>/user/warehouse/"
+);
+```
+
+More catalog types and configures can be seen in [Paimon catalog](https://docs.starrocks.io/docs/data_source/catalog/paimon_catalog/).
+
+## Query
+Suppose there already exists a database named `test_db` and a table named `test_tbl` in `paimon_catalog`,
+you can query this table using the following SQL:
+```sql
+SELECT * FROM paimon_catalog.test_db.test_tbl;
+```
+
+## Query System Tables
+
+You can access all kinds of Paimon system tables by StarRocks. For example, you can read the `ro`
+(read-optimized) system table to improve reading performance of primary-key tables.
+
+```sql
+SELECT * FROM paimon_catalog.test_db.test_tbl$ro;
+```
+
+For another example, you can query partition files of the table using the following SQL:
+
+```sql
+SELECT * FROM paimon_catalog.test_db.partition_tbl$partitions;
+/*
++-----------+--------------+--------------------+------------+----------------------------+
+| partition | record_count | file_size_in_bytes | file_count | last_update_time |
++-----------+--------------+--------------------+------------+----------------------------+
+| [1] | 1 | 645 | 1 | 2024-01-01 00:00:00.000000 |
++-----------+--------------+--------------------+------------+----------------------------+
+*/
+```
+
+## StarRocks to Paimon type mapping
+
+This section lists all supported type conversion between StarRocks and Paimon.
+All StarRocks’s data types can be found in this doc [StarRocks Data type overview](https://docs.starrocks.io/docs/sql-reference/data-types/data-type-list/).
+
+<table class="table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 10%">StarRocks Data Type</th>
+ <th class="text-left" style="width: 10%">Paimon Data Type</th>
+ <th class="text-left" style="width: 5%">Atomic Type</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td><code>STRUCT</code></td>
+ <td><code>RowType</code></td>
+ <td>false</td>
+ </tr>
+ <tr>
+ <td><code>MAP</code></td>
+ <td><code>MapType</code></td>
+ <td>false</td>
+ </tr>
+ <tr>
+ <td><code>ARRAY</code></td>
+ <td><code>ArrayType</code></td>
+ <td>false</td>
+ </tr>
+ <tr>
+ <td><code>BOOLEAN</code></td>
+ <td><code>BooleanType</code></td>
+ <td>true</td>
+ </tr>
+ <tr>
+ <td><code>TINYINT</code></td>
+ <td><code>TinyIntType</code></td>
+ <td>true</td>
+ </tr>
+ <tr>
+ <td><code>SMALLINT</code></td>
+ <td><code>SmallIntType</code></td>
+ <td>true</td>
+ </tr>
+ <tr>
+ <td><code>INT</code></td>
+ <td><code>IntType</code></td>
+ <td>true</td>
+ </tr>
+ <tr>
+ <td><code>BIGINT</code></td>
+ <td><code>BigIntType</code></td>
+ <td>true</td>
+ </tr>
+ <tr>
+ <td><code>FLOAT</code></td>
+ <td><code>FloatType</code></td>
+ <td>true</td>
+ </tr>
+ <tr>
+ <td><code>DOUBLE</code></td>
+ <td><code>DoubleType</code></td>
+ <td>true</td>
+ </tr>
+ <tr>
+ <td><code>CHAR(length)</code></td>
+ <td><code>CharType(length)</code></td>
+ <td>true</td>
+ </tr>
+ <tr>
+ <td><code>VARCHAR(MAX_VARCHAR_LENGTH)</code></td>
+ <td><code>VarCharType(VarCharType.MAX_LENGTH)</code></td>
+ <td>true</td>
+ </tr>
+ <tr>
+ <td><code>VARCHAR(length)</code></td>
+ <td><code>VarCharType(length), length is less than VarCharType.MAX_LENGTH</code></td>
+ <td>true</td>
+ </tr>
+ <tr>
+ <td><code>DATE</code></td>
+ <td><code>DateType</code></td>
+ <td>true</td>
+ </tr>
+ <tr>
+ <td><code>DATETIME</code></td>
+ <td><code>TimestampType</code></td>
+ <td>true</td>
+ </tr>
+ <tr>
+ <td><code>DECIMAL(precision, scale)</code></td>
+ <td><code>DecimalType(precision, scale)</code></td>
+ <td>true</td>
+ </tr>
+ <tr>
+ <td><code>VARBINARY(length)</code></td>
+ <td><code>VarBinaryType(length)</code></td>
+ <td>true</td>
+ </tr>
+ <tr>
+ <td><code>DATETIME</code></td>
+ <td><code>LocalZonedTimestampType</code></td>
+ <td>true</td>
+ </tr>
+ </tbody>
+</table>
diff --git a/docs/content/maintenance/system-tables.md b/docs/content/maintenance/system-tables.md
index ffef040e9..b4f0168a3 100644
--- a/docs/content/maintenance/system-tables.md
+++ b/docs/content/maintenance/system-tables.md
@@ -30,7 +30,7 @@ under the License.
Table specified system tables contain metadata and information about each table, such as the snapshots created and the options in use. Users can access system tables with batch queries.
-Currently, Flink, Spark and Trino supports querying system tables.
+Currently, Flink, Spark, Trino and StarRocks support querying system tables.
In some cases, the table name needs to be enclosed with back quotes to avoid syntax parsing conflicts, for example triple access mode:
```sql
(paimon) 16/31: [bundle] add source code to the source jar for paimon-bundle (#3333)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit ee3cd40a13fa39ec7c8ce7332bfd090ce85f9b6f
Author: wgcn <10...@qq.com>
AuthorDate: Wed May 15 14:54:52 2024 +0800
[bundle] add source code to the source jar for paimon-bundle (#3333)
---
paimon-bundle/pom.xml | 66 +++++++++++++++++++++++++++++
paimon-bundle/src/assembly/src-assembly.xml | 37 ++++++++++++++++
2 files changed, 103 insertions(+)
diff --git a/paimon-bundle/pom.xml b/paimon-bundle/pom.xml
index 6f791bd14..003ffe98b 100644
--- a/paimon-bundle/pom.xml
+++ b/paimon-bundle/pom.xml
@@ -161,4 +161,70 @@ under the License.
</plugin>
</plugins>
</build>
+
+ <profiles>
+ <profile>
+ <id>docs-and-source</id>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-dependency-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>unpack-dependencies</id>
+ <phase>generate-sources</phase>
+ <goals>
+ <goal>unpack-dependencies</goal>
+ </goals>
+ <configuration>
+ <includeGroupIds>org.apache.paimon</includeGroupIds>
+ <classifier>sources</classifier>
+ <outputDirectory>${project.build.directory}/dependency-sources</outputDirectory>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-assembly-plugin</artifactId>
+ <version>3.4.2</version>
+ <executions>
+ <execution>
+ <id>make-sources-jar</id>
+ <phase>package</phase>
+ <goals>
+ <goal>single</goal>
+ </goals>
+ <configuration>
+ <descriptors>
+ <descriptor>src/assembly/src-assembly.xml</descriptor>
+ </descriptors>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-javadoc-plugin</artifactId>
+ <configuration>
+ <quiet>true</quiet>
+ </configuration>
+ <executions>
+ <execution>
+ <id>attach-javadocs</id>
+ <goals>
+ <goal>jar</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
+
+ </profiles>
</project>
diff --git a/paimon-bundle/src/assembly/src-assembly.xml b/paimon-bundle/src/assembly/src-assembly.xml
new file mode 100644
index 000000000..93b24109b
--- /dev/null
+++ b/paimon-bundle/src/assembly/src-assembly.xml
@@ -0,0 +1,37 @@
+<!--
+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.
+-->
+<assembly
+ xmlns="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0 http://maven.apache.org/xsd/assembly-1.1.0.xsd">
+ <id>sources</id>
+ <formats>
+ <format>jar</format>
+ </formats>
+ <includeBaseDirectory>false</includeBaseDirectory>
+ <fileSets>
+ <fileSet>
+ <directory>${project.build.directory}/dependency-sources</directory>
+ <outputDirectory>/</outputDirectory>
+ <includes>
+ <include>**/*</include>
+ </includes>
+ </fileSet>
+ </fileSets>
+</assembly>
\ No newline at end of file
(paimon) 30/31: [spark] SparkHilbertUDF hilbertCurvePosBytes classnotfound fix (#3415)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit d78658a187a303c569d6c4f45df2894fcb106dfd
Author: xuzifu666 <12...@qq.com>
AuthorDate: Wed May 29 10:28:19 2024 +0800
[spark] SparkHilbertUDF hilbertCurvePosBytes classnotfound fix (#3415)
---
.../org/apache/paimon/sort/hilbert/HilbertIndexer.java | 2 +-
paimon-spark/paimon-spark-common/pom.xml | 5 -----
.../java/org/apache/paimon/spark/sort/SparkHilbertUDF.java | 14 ++++----------
3 files changed, 5 insertions(+), 16 deletions(-)
diff --git a/paimon-common/src/main/java/org/apache/paimon/sort/hilbert/HilbertIndexer.java b/paimon-common/src/main/java/org/apache/paimon/sort/hilbert/HilbertIndexer.java
index e0e55795f..b75227309 100644
--- a/paimon-common/src/main/java/org/apache/paimon/sort/hilbert/HilbertIndexer.java
+++ b/paimon-common/src/main/java/org/apache/paimon/sort/hilbert/HilbertIndexer.java
@@ -295,7 +295,7 @@ public class HilbertIndexer implements Serializable {
}
}
- private byte[] hilbertCurvePosBytes(Long[] points) {
+ public static byte[] hilbertCurvePosBytes(Long[] points) {
long[] data = Arrays.stream(points).mapToLong(Long::longValue).toArray();
HilbertCurve hilbertCurve = HilbertCurve.bits(BITS_NUM).dimensions(points.length);
BigInteger index = hilbertCurve.index(data);
diff --git a/paimon-spark/paimon-spark-common/pom.xml b/paimon-spark/paimon-spark-common/pom.xml
index e435174f8..29deec4af 100644
--- a/paimon-spark/paimon-spark-common/pom.xml
+++ b/paimon-spark/paimon-spark-common/pom.xml
@@ -53,11 +53,6 @@ under the License.
<artifactId>scala-compiler</artifactId>
<version>${scala.version}</version>
</dependency>
- <dependency>
- <groupId>com.github.davidmoten</groupId>
- <artifactId>hilbert-curve</artifactId>
- <version>0.2.2</version>
- </dependency>
<dependency>
<groupId>org.apache.spark</groupId>
diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/sort/SparkHilbertUDF.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/sort/SparkHilbertUDF.java
index ed2a1bab0..f9334c097 100644
--- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/sort/SparkHilbertUDF.java
+++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/sort/SparkHilbertUDF.java
@@ -18,6 +18,7 @@
package org.apache.paimon.spark.sort;
+import org.apache.paimon.sort.hilbert.HilbertIndexer;
import org.apache.paimon.utils.ConvertBinaryUtil;
import org.apache.spark.sql.Column;
@@ -37,14 +38,11 @@ import org.apache.spark.sql.types.LongType;
import org.apache.spark.sql.types.ShortType;
import org.apache.spark.sql.types.StringType;
import org.apache.spark.sql.types.TimestampType;
-import org.davidmoten.hilbert.HilbertCurve;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.Serializable;
import java.math.BigDecimal;
-import java.math.BigInteger;
-import java.util.List;
import scala.collection.JavaConverters;
import scala.collection.Seq;
@@ -53,8 +51,6 @@ import scala.collection.Seq;
public class SparkHilbertUDF implements Serializable {
private static final long PRIMITIVE_EMPTY = Long.MAX_VALUE;
- private static final int BITS_NUM = 63;
-
SparkHilbertUDF() {}
private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
@@ -62,11 +58,9 @@ public class SparkHilbertUDF implements Serializable {
}
byte[] hilbertCurvePosBytes(Seq<Long> points) {
- List<Long> longs = JavaConverters.seqAsJavaList(points);
- long[] data = longs.stream().mapToLong(Long::longValue).toArray();
- HilbertCurve hilbertCurve = HilbertCurve.bits(BITS_NUM).dimensions(points.size());
- BigInteger index = hilbertCurve.index(data);
- return ConvertBinaryUtil.paddingToNByte(index.toByteArray(), BITS_NUM);
+ Long[] longs = JavaConverters.seqAsJavaList(points).stream().toArray(Long[]::new);
+ byte[] bytes = HilbertIndexer.hilbertCurvePosBytes(longs);
+ return bytes;
}
private UserDefinedFunction tinyToOrderedLongUDF() {
(paimon) 28/31: [doc] Fix typo in mongo-cdc.md (#3413)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 42299afdc053553c2fc9decd18c6ec22c3cfe577
Author: LJ <97...@users.noreply.github.com>
AuthorDate: Tue May 28 20:00:42 2024 +0800
[doc] Fix typo in mongo-cdc.md (#3413)
---
docs/content/flink/cdc-ingestion/mongo-cdc.md | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/docs/content/flink/cdc-ingestion/mongo-cdc.md b/docs/content/flink/cdc-ingestion/mongo-cdc.md
index 64cd5b431..17c70ba40 100644
--- a/docs/content/flink/cdc-ingestion/mongo-cdc.md
+++ b/docs/content/flink/cdc-ingestion/mongo-cdc.md
@@ -119,13 +119,13 @@ Path Examples
{{< generated/mongodb_path_example >}}
-2. The synchronized table is required to have its primary key set as `_id`.
+1. The synchronized table is required to have its primary key set as `_id`.
This is because MongoDB's change events are recorded before updates in messages.
Consequently, we can only convert them into Flink's UPSERT change log stream.
The upstart stream demands a unique key, which is why we must declare `_id` as the primary key.
Declaring other columns as primary keys is not feasible, as delete operations only encompass the _id and sharding key, excluding other keys and values.
-3. MongoDB Change Streams are designed to return simple JSON documents without any data type definitions. This is because MongoDB is a document-oriented database, and one of its core features is the dynamic schema, where documents can contain different fields, and the data types of fields can be flexible. Therefore, the absence of data type definitions in Change Streams is to maintain this flexibility and extensibility.
+2. MongoDB Change Streams are designed to return simple JSON documents without any data type definitions. This is because MongoDB is a document-oriented database, and one of its core features is the dynamic schema, where documents can contain different fields, and the data types of fields can be flexible. Therefore, the absence of data type definitions in Change Streams is to maintain this flexibility and extensibility.
For this reason, we have set all field data types for synchronizing MongoDB to Paimon as String to address the issue of not being able to obtain data types.
If the Paimon table you specify does not exist, this action will automatically create the table. Its schema will be derived from MongoDB collection.
(paimon) 17/31: [flink] Respect waitCompaction in FlinkCdcMultiTableSink (#3337)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit b1944274e4ecab1b1c4143566f3bbd0c86d19c79
Author: YeJunHao <41...@users.noreply.github.com>
AuthorDate: Wed May 15 18:29:55 2024 +0800
[flink] Respect waitCompaction in FlinkCdcMultiTableSink (#3337)
---
.../java/org/apache/paimon/flink/sink/cdc/FlinkCdcMultiTableSink.java | 4 +++-
1 file changed, 3 insertions(+), 1 deletion(-)
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcMultiTableSink.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcMultiTableSink.java
index cff31087b..4254cb6b0 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcMultiTableSink.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcMultiTableSink.java
@@ -19,6 +19,7 @@
package org.apache.paimon.flink.sink.cdc;
import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.flink.FlinkConnectorOptions;
import org.apache.paimon.flink.VersionedSerializerWrapper;
import org.apache.paimon.flink.sink.CommittableStateManager;
import org.apache.paimon.flink.sink.Committer;
@@ -83,7 +84,8 @@ public class FlinkCdcMultiTableSink implements Serializable {
state,
ioManager,
isOverwrite,
- false,
+ FlinkConnectorOptions.prepareCommitWaitCompaction(
+ table.coreOptions().toConfiguration()),
true,
memoryPoolFactory,
metricGroup);
(paimon) 09/31: [core] FileStoreWrite.State should get max sequence number from RecordWriter instead of from data files (#3327)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 92e2f00f82edaad2340634eb6180e08d97da65bf
Author: yuzelin <33...@users.noreply.github.com>
AuthorDate: Mon May 13 15:34:04 2024 +0800
[core] FileStoreWrite.State should get max sequence number from RecordWriter instead of from data files (#3327)
---
.../org/apache/paimon/append/AppendOnlyWriter.java | 5 +++
.../apache/paimon/mergetree/MergeTreeWriter.java | 5 +++
.../paimon/operation/AbstractFileStoreWrite.java | 6 ++++
.../paimon/operation/AppendOnlyFileStoreWrite.java | 6 ++--
.../apache/paimon/operation/FileStoreWrite.java | 6 +++-
.../paimon/operation/KeyValueFileStoreWrite.java | 4 +--
.../java/org/apache/paimon/utils/RecordWriter.java | 3 ++
.../cdc/mysql/MySqlSyncTableActionITCase.java | 42 ++++++++++++++++++++++
.../src/test/resources/mysql/sync_table_setup.sql | 11 ++++++
9 files changed, 81 insertions(+), 7 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyWriter.java b/paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyWriter.java
index 0ae21e6b2..52f6241a9 100644
--- a/paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyWriter.java
+++ b/paimon-core/src/main/java/org/apache/paimon/append/AppendOnlyWriter.java
@@ -173,6 +173,11 @@ public class AppendOnlyWriter implements RecordWriter<InternalRow>, MemoryOwner
return compactManager.allFiles();
}
+ @Override
+ public long maxSequenceNumber() {
+ return seqNumCounter.getValue() - 1;
+ }
+
@Override
public CommitIncrement prepareCommit(boolean waitCompaction) throws Exception {
flush(false, false);
diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/MergeTreeWriter.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/MergeTreeWriter.java
index 17673fc06..6becbf731 100644
--- a/paimon-core/src/main/java/org/apache/paimon/mergetree/MergeTreeWriter.java
+++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/MergeTreeWriter.java
@@ -182,6 +182,11 @@ public class MergeTreeWriter implements RecordWriter<KeyValue>, MemoryOwner {
return compactManager.allFiles();
}
+ @Override
+ public long maxSequenceNumber() {
+ return newSequenceNumber - 1;
+ }
+
@Override
public long memoryOccupancy() {
return writeBuffer.memoryOccupancy();
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreWrite.java b/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreWrite.java
index 30ee9c6c1..e4ef21b55 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreWrite.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreWrite.java
@@ -52,6 +52,8 @@ import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import static org.apache.paimon.io.DataFileMeta.getMaxSequenceNumber;
+
/**
* Base {@link FileStoreWrite} implementation.
*
@@ -297,6 +299,7 @@ public abstract class AbstractFileStoreWrite<T> implements FileStoreWrite<T> {
writerContainer.baseSnapshotId,
writerContainer.lastModifiedCommitIdentifier,
dataFiles,
+ writerContainer.writer.maxSequenceNumber(),
writerContainer.indexMaintainer,
writerContainer.deletionVectorsMaintainer,
increment));
@@ -317,6 +320,7 @@ public abstract class AbstractFileStoreWrite<T> implements FileStoreWrite<T> {
state.partition,
state.bucket,
state.dataFiles,
+ state.maxSequenceNumber,
state.commitIncrement,
compactExecutor(),
state.deletionVectorsMaintainer);
@@ -382,6 +386,7 @@ public abstract class AbstractFileStoreWrite<T> implements FileStoreWrite<T> {
partition.copy(),
bucket,
restoreFiles,
+ getMaxSequenceNumber(restoreFiles),
null,
compactExecutor(),
deletionVectorsMaintainer);
@@ -432,6 +437,7 @@ public abstract class AbstractFileStoreWrite<T> implements FileStoreWrite<T> {
BinaryRow partition,
int bucket,
List<DataFileMeta> restoreFiles,
+ long restoredMaxSeqNumber,
@Nullable CommitIncrement restoreIncrement,
ExecutorService compactExecutor,
@Nullable DeletionVectorsMaintainer deletionVectorsMaintainer);
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/AppendOnlyFileStoreWrite.java b/paimon-core/src/main/java/org/apache/paimon/operation/AppendOnlyFileStoreWrite.java
index bb3074ac7..e7ae19919 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/AppendOnlyFileStoreWrite.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/AppendOnlyFileStoreWrite.java
@@ -54,8 +54,6 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
-import static org.apache.paimon.io.DataFileMeta.getMaxSequenceNumber;
-
/** {@link FileStoreWrite} for {@link AppendOnlyFileStore}. */
public class AppendOnlyFileStoreWrite extends MemoryFileStoreWrite<InternalRow> {
@@ -119,12 +117,12 @@ public class AppendOnlyFileStoreWrite extends MemoryFileStoreWrite<InternalRow>
BinaryRow partition,
int bucket,
List<DataFileMeta> restoredFiles,
+ long restoredMaxSeqNumber,
@Nullable CommitIncrement restoreIncrement,
ExecutorService compactExecutor,
@Nullable DeletionVectorsMaintainer ignore) {
// let writer and compact manager hold the same reference
// and make restore files mutable to update
- long maxSequenceNumber = getMaxSequenceNumber(restoredFiles);
DataFilePathFactory factory = pathFactory.createDataFilePathFactory(partition, bucket);
CompactManager compactManager =
skipCompaction
@@ -147,7 +145,7 @@ public class AppendOnlyFileStoreWrite extends MemoryFileStoreWrite<InternalRow>
fileFormat,
targetFileSize,
rowType,
- maxSequenceNumber,
+ restoredMaxSeqNumber,
compactManager,
bucketReader(partition, bucket),
commitForceCompact,
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreWrite.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreWrite.java
index 1391b6916..7f8fee45f 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreWrite.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreWrite.java
@@ -146,6 +146,7 @@ public interface FileStoreWrite<T> extends Restorable<List<FileStoreWrite.State<
protected final long baseSnapshotId;
protected final long lastModifiedCommitIdentifier;
protected final List<DataFileMeta> dataFiles;
+ protected final long maxSequenceNumber;
@Nullable protected final IndexMaintainer<T> indexMaintainer;
@Nullable protected final DeletionVectorsMaintainer deletionVectorsMaintainer;
protected final CommitIncrement commitIncrement;
@@ -156,6 +157,7 @@ public interface FileStoreWrite<T> extends Restorable<List<FileStoreWrite.State<
long baseSnapshotId,
long lastModifiedCommitIdentifier,
Collection<DataFileMeta> dataFiles,
+ long maxSequenceNumber,
@Nullable IndexMaintainer<T> indexMaintainer,
@Nullable DeletionVectorsMaintainer deletionVectorsMaintainer,
CommitIncrement commitIncrement) {
@@ -164,6 +166,7 @@ public interface FileStoreWrite<T> extends Restorable<List<FileStoreWrite.State<
this.baseSnapshotId = baseSnapshotId;
this.lastModifiedCommitIdentifier = lastModifiedCommitIdentifier;
this.dataFiles = new ArrayList<>(dataFiles);
+ this.maxSequenceNumber = maxSequenceNumber;
this.indexMaintainer = indexMaintainer;
this.deletionVectorsMaintainer = deletionVectorsMaintainer;
this.commitIncrement = commitIncrement;
@@ -172,12 +175,13 @@ public interface FileStoreWrite<T> extends Restorable<List<FileStoreWrite.State<
@Override
public String toString() {
return String.format(
- "{%s, %d, %d, %d, %s, %s, %s, %s}",
+ "{%s, %d, %d, %d, %s, %d, %s, %s, %s}",
partition,
bucket,
baseSnapshotId,
lastModifiedCommitIdentifier,
dataFiles,
+ maxSequenceNumber,
indexMaintainer,
deletionVectorsMaintainer,
commitIncrement);
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java b/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java
index 314019a60..b7702ef8a 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java
@@ -84,7 +84,6 @@ import java.util.function.Supplier;
import static org.apache.paimon.CoreOptions.ChangelogProducer.FULL_COMPACTION;
import static org.apache.paimon.CoreOptions.MergeEngine.DEDUPLICATE;
import static org.apache.paimon.CoreOptions.MergeEngine.FIRST_ROW;
-import static org.apache.paimon.io.DataFileMeta.getMaxSequenceNumber;
import static org.apache.paimon.lookup.LookupStoreFactory.bfGenerator;
/** {@link FileStoreWrite} for {@link KeyValueFileStore}. */
@@ -168,6 +167,7 @@ public class KeyValueFileStoreWrite extends MemoryFileStoreWrite<KeyValue> {
BinaryRow partition,
int bucket,
List<DataFileMeta> restoreFiles,
+ long restoredMaxSeqNumber,
@Nullable CommitIncrement restoreIncrement,
ExecutorService compactExecutor,
@Nullable DeletionVectorsMaintainer dvMaintainer) {
@@ -204,7 +204,7 @@ public class KeyValueFileStoreWrite extends MemoryFileStoreWrite<KeyValue> {
options.spillCompression(),
ioManager,
compactManager,
- getMaxSequenceNumber(restoreFiles),
+ restoredMaxSeqNumber,
keyComparator,
mfFactory.create(),
writerFactory,
diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/RecordWriter.java b/paimon-core/src/main/java/org/apache/paimon/utils/RecordWriter.java
index d2df1be46..5795290f1 100644
--- a/paimon-core/src/main/java/org/apache/paimon/utils/RecordWriter.java
+++ b/paimon-core/src/main/java/org/apache/paimon/utils/RecordWriter.java
@@ -53,6 +53,9 @@ public interface RecordWriter<T> {
/** Get all data files maintained by this writer. */
Collection<DataFileMeta> dataFiles();
+ /** Get max sequence number of records written by this writer. */
+ long maxSequenceNumber();
+
/**
* Prepare for a commit.
*
diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableActionITCase.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableActionITCase.java
index 58ed2098e..d4f847b0a 100644
--- a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableActionITCase.java
+++ b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/mysql/MySqlSyncTableActionITCase.java
@@ -18,6 +18,7 @@
package org.apache.paimon.flink.action.cdc.mysql;
+import org.apache.paimon.CoreOptions;
import org.apache.paimon.catalog.FileSystemCatalogOptions;
import org.apache.paimon.options.CatalogOptions;
import org.apache.paimon.schema.SchemaChange;
@@ -1364,4 +1365,45 @@ public class MySqlSyncTableActionITCase extends MySqlActionITCaseBase {
assertThat(actual.get("c1").description()).isEqualTo("c1 comment");
assertThat(actual.get("c2").description()).isEqualTo("c2 comment");
}
+
+ @Test
+ @Timeout(60)
+ public void testWriteOnlyAndSchemaEvolution() throws Exception {
+ Map<String, String> mySqlConfig = getBasicMySqlConfig();
+ mySqlConfig.put("database-name", "write_only_and_schema_evolution");
+ mySqlConfig.put("table-name", "t");
+
+ Map<String, String> tableConfig = getBasicTableConfig();
+ tableConfig.put(CoreOptions.WRITE_ONLY.key(), "true");
+
+ MySqlSyncTableAction action =
+ syncTableActionBuilder(mySqlConfig).withTableConfig(tableConfig).build();
+
+ runActionWithDefaultEnv(action);
+ FileStoreTable table = getFileStoreTable();
+
+ try (Statement statement = getStatement()) {
+ statement.executeUpdate("USE write_only_and_schema_evolution");
+ statement.executeUpdate("INSERT INTO t VALUES (1, 'one'), (2, 'two')");
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.INT().notNull(), DataTypes.VARCHAR(10)},
+ new String[] {"k", "v1"});
+ List<String> primaryKeys = Collections.singletonList("k");
+ List<String> expected = Arrays.asList("+I[1, one]", "+I[2, two]");
+ waitForResult(expected, table, rowType, primaryKeys);
+
+ statement.executeUpdate("ALTER TABLE t ADD COLUMN v2 INT");
+ statement.executeUpdate("UPDATE t SET v2 = 1 WHERE k = 1");
+
+ rowType =
+ RowType.of(
+ new DataType[] {
+ DataTypes.INT().notNull(), DataTypes.VARCHAR(10), DataTypes.INT()
+ },
+ new String[] {"k", "v1", "v2"});
+ expected = Arrays.asList("+I[1, one, 1]", "+I[2, two, NULL]");
+ waitForResult(expected, table, rowType, primaryKeys);
+ }
+ }
}
diff --git a/paimon-flink/paimon-flink-cdc/src/test/resources/mysql/sync_table_setup.sql b/paimon-flink/paimon-flink-cdc/src/test/resources/mysql/sync_table_setup.sql
index 5cf9cc1d9..676185fb9 100644
--- a/paimon-flink/paimon-flink-cdc/src/test/resources/mysql/sync_table_setup.sql
+++ b/paimon-flink/paimon-flink-cdc/src/test/resources/mysql/sync_table_setup.sql
@@ -420,3 +420,14 @@ USE invalid_alter_bucket;
CREATE TABLE t (
k INT PRIMARY KEY
);
+
+-- ################################################################################
+-- testInvalidAlterBucket
+-- ################################################################################
+
+CREATE DATABASE write_only_and_schema_evolution;
+USE write_only_and_schema_evolution;
+CREATE TABLE t (
+ k INT PRIMARY KEY,
+ v1 VARCHAR(10)
+);
(paimon) 07/31: [doc] Update version and link of Doris and Trino (#3322)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit a177de33a8e07010139a9ee116268ccac40a752f
Author: 911432 <30...@users.noreply.github.com>
AuthorDate: Mon May 13 11:54:12 2024 +0900
[doc] Update version and link of Doris and Trino (#3322)
---
docs/content/engines/overview.md | 9 ++++-----
1 file changed, 4 insertions(+), 5 deletions(-)
diff --git a/docs/content/engines/overview.md b/docs/content/engines/overview.md
index 6db9dbbf7..19f3bf1c1 100644
--- a/docs/content/engines/overview.md
+++ b/docs/content/engines/overview.md
@@ -28,16 +28,15 @@ under the License.
## Compatibility Matrix
-| Engine | Version | Batch Read | Batch Write | Create Table | Alter Table | Streaming Write | Streaming Read | Batch Overwrite | DELETE & UPDATE | MERGE INTO |
-|:-------------------------------------------------------------------------------:|:-------------:|:-----------:|:------------:|:------------:|:------------:|:----------------:|:--------------:|:----------------:|:----------------:|:-----------:|
+| Engine | Version | Batch Read | Batch Write | Create Table | Alter Table | Streaming Write | Streaming Read | Batch Overwrite | DELETE & UPDATE | MERGE INTO |
+|:-------------------------------------------------------------------------------:|:-------------:|:-----------:|:-------------:|:-------------:|:-------------:|:----------------:|:----------------:|:----------------:|:------------------:|:-----------:|
| Flink | 1.15 - 1.19 | ✅ | ✅ | ✅ | ✅(1.17+) | ✅ | ✅ | ✅ | ✅(1.17+) | ❌ |
| Spark | 3.1 - 3.5 | ✅ | ✅(3.3+) | ✅ | ✅ | ✅(3.3+) | ✅(3.3+) | ✅(3.3+) | ✅(3.2+) | ✅(3.2+) |
| Hive | 2.1 - 3.1 | ✅ | ✅ | ✅ | ❌ | ❌ | ❌ | ❌ | ❌ | ❌ |
-| Trino | 420 - 426 | ✅ | ❌ | ❌ | ❌ | ❌ | ❌ | ❌ | ❌ | ❌ |
-| Trino | 427 - 439 | ✅ | ❌ | ✅ | ✅ | ❌ | ❌ | ❌ | ❌ | ❌ |
+| Trino | 420 - 439 | ✅ | ✅(427+) | ✅(427+) | ✅(427+) | ❌ | ❌ | ❌ | ❌ | ❌ |
| Presto | 0.236 - 0.280 | ✅ | ❌ | ✅ | ✅ | ❌ | ❌ | ❌ | ❌ | ❌ |
| [StarRocks](https://docs.starrocks.io/docs/data_source/catalog/paimon_catalog/) | 3.1+ | ✅ | ❌ | ❌ | ❌ | ❌ | ❌ | ❌ | ❌ | ❌ |
-| [Doris](https://doris.apache.org/docs/lakehouse/multi-catalog/paimon/) | 2.0+ | ✅ | ❌ | ❌ | ❌ | ❌ | ❌ | ❌ | ❌ | ❌ |
+| [Doris](https://doris.apache.org/docs/lakehouse/datalake-analytics/paimon) | 2.0.6+ | ✅ | ❌ | ❌ | ❌ | ❌ | ❌ | ❌ | ❌ | ❌ |
## Streaming Engines
(paimon) 27/31: [doc] Remove outdate roadmap page
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 67db033e8ac505bb4f3bb5054ff7b9c51a869a4c
Author: Jingsong <ji...@gmail.com>
AuthorDate: Mon May 27 11:06:53 2024 +0800
[doc] Remove outdate roadmap page
---
docs/content/project/roadmap.md | 48 -----------------------------------------
1 file changed, 48 deletions(-)
diff --git a/docs/content/project/roadmap.md b/docs/content/project/roadmap.md
deleted file mode 100644
index 80edb3568..000000000
--- a/docs/content/project/roadmap.md
+++ /dev/null
@@ -1,48 +0,0 @@
----
-title: "Roadmap"
-weight: 1
-type: docs
-aliases:
-- /project/roadmap.html
----
-<!--
-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.
--->
-
-# Roadmap
-
-Paimon's long-term goal is to become the better data lake platform for building the Streaming Lakehouse. Paimon will
-invest in real-time, ecology, and data warehouse integrity for a long time.
-
-If you have other requirements, please contact us.
-
-## What’s Next?
-
-### Core
-
-- Foreign Key Join Partial-Update
-- Append table supports batch / streaming z-order sort
-- Supports cross partition update
-
-### Compute Engines
-
-- More management via Flink/Spark `CALL` procedures
-- Flink Sink supports at-least-once / unaligned checkpoint
-- Flink Whole database compaction Job
-- Public Spark Schema Evolution Pipeline
-- Spark supports Dynamic Partition overwrite
(paimon) 20/31: [spark] Fix drop database cascade when database with view with SparkGenericCatalog (#3347)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 0b63cb8a965b6aeec27ef9fb5f033da6755755f7
Author: Zouxxyy <zo...@alibaba-inc.com>
AuthorDate: Mon May 20 10:52:53 2024 +0800
[spark] Fix drop database cascade when database with view with SparkGenericCatalog (#3347)
---
.../main/java/org/apache/paimon/spark/SparkGenericCatalog.java | 9 ++++++++-
.../test/scala/org/apache/paimon/spark/PaimonSparkTestBase.scala | 5 +++++
.../org/apache/paimon/spark/sql/DDLWithHiveCatalogTestBase.scala | 8 ++++++++
3 files changed, 21 insertions(+), 1 deletion(-)
diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkGenericCatalog.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkGenericCatalog.java
index b947523fe..9576379f8 100644
--- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkGenericCatalog.java
+++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkGenericCatalog.java
@@ -134,7 +134,14 @@ public class SparkGenericCatalog extends SparkBaseCatalog implements CatalogExte
throws NoSuchNamespaceException, NonEmptyNamespaceException {
if (namespace.length == 1 && namespaceExists(namespace) && cascade) {
for (Identifier table : listTables(namespace)) {
- dropTable(table);
+ try {
+ dropTable(table);
+ } catch (Exception e) {
+ LOG.warn(
+ "Failed to drop {}, fallback to use sessionCatalog to drop, for {}",
+ table,
+ e.getMessage());
+ }
}
}
return asNamespaceCatalog().dropNamespace(namespace, cascade);
diff --git a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/PaimonSparkTestBase.scala b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/PaimonSparkTestBase.scala
index 6ccf08016..814001bc9 100644
--- a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/PaimonSparkTestBase.scala
+++ b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/PaimonSparkTestBase.scala
@@ -84,6 +84,11 @@ class PaimonSparkTestBase
}
}
+ protected def reset(): Unit = {
+ afterAll()
+ beforeAll()
+ }
+
/** Default is paimon catalog */
override protected def beforeEach(): Unit = {
super.beforeAll()
diff --git a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/DDLWithHiveCatalogTestBase.scala b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/DDLWithHiveCatalogTestBase.scala
index c03f61bf6..c542633dc 100644
--- a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/DDLWithHiveCatalogTestBase.scala
+++ b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/DDLWithHiveCatalogTestBase.scala
@@ -99,7 +99,9 @@ abstract class DDLWithHiveCatalogTestBase extends PaimonHiveTestBase {
}
}
+ // Since we created a new sparkContext, we need to stop it and reset the default sparkContext
reusedSpark.stop()
+ reset()
}
test("Paimon DDL with hive catalog: drop database cascade which contains paimon table") {
@@ -111,6 +113,12 @@ abstract class DDLWithHiveCatalogTestBase extends PaimonHiveTestBase {
spark.sql(s"CREATE DATABASE paimon_db")
spark.sql(s"USE paimon_db")
spark.sql(s"CREATE TABLE paimon_tbl (id int, name string, dt string) using paimon")
+ // Currently, only spark_catalog supports create other table or view
+ if (catalogName.equals(sparkCatalogName)) {
+ spark.sql(s"CREATE TABLE parquet_tbl (id int, name string, dt string) using parquet")
+ spark.sql(s"CREATE VIEW parquet_tbl_view AS SELECT * FROM parquet_tbl")
+ spark.sql(s"CREATE VIEW paimon_tbl_view AS SELECT * FROM paimon_tbl")
+ }
spark.sql(s"USE default")
spark.sql(s"DROP DATABASE paimon_db CASCADE")
}
(paimon) 02/31: [minor] Fix exception message in FileStoreCommitImpl
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit ad371d5ca7bd9c7f27e12a0dc6115e435296a77e
Author: Jingsong <ji...@gmail.com>
AuthorDate: Mon May 6 18:18:11 2024 +0800
[minor] Fix exception message in FileStoreCommitImpl
---
.../src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
index e389a471c..be5255a36 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
@@ -1090,7 +1090,7 @@ public class FileStoreCommitImpl implements FileStoreCommit {
" You'll probably see different base commit user and current commit user below.",
" You can use "
+ "https://paimon.apache.org/docs/master/maintenance/dedicated-compaction#dedicated-compaction-job"
- + "to support multiple writing.",
+ + " to support multiple writing.",
"3. You're recovering from an old savepoint, or you're creating multiple jobs from a savepoint.",
" The job will fail continuously in this scenario to protect metadata from corruption.",
" You can either recover from the latest savepoint, "
(paimon) 05/31: [doc] Reorg Append table pages
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit d83a06255d2ce8f9871ef3ac72457abfc50b5753
Author: Jingsong <ji...@gmail.com>
AuthorDate: Sat May 11 14:12:00 2024 +0800
[doc] Reorg Append table pages
---
.../{append-queue-table.md => append-queue.md} | 48 ++++-----
docs/content/append-table/append-scalable-table.md | 114 ---------------------
docs/content/append-table/append-table.md | 64 ++++++++++++
docs/content/append-table/overview.md | 36 -------
docs/content/learn-paimon/understand-files.md | 2 +-
docs/content/maintenance/dedicated-compaction.md | 2 +-
docs/content/migration/migration-from-hive.md | 2 +-
docs/static/img/for-scalable.png | Bin 1331757 -> 0 bytes
8 files changed, 89 insertions(+), 179 deletions(-)
diff --git a/docs/content/append-table/append-queue-table.md b/docs/content/append-table/append-queue.md
similarity index 92%
rename from docs/content/append-table/append-queue-table.md
rename to docs/content/append-table/append-queue.md
index 3d3e3f22e..07bd7d980 100644
--- a/docs/content/append-table/append-queue-table.md
+++ b/docs/content/append-table/append-queue.md
@@ -1,9 +1,9 @@
---
-title: "Append Queue Table"
+title: "Append Queue"
weight: 3
type: docs
aliases:
-- /append-table/append-queue-table.html
+- /append-table/append-queue.html
---
<!--
Licensed to the Apache Software Foundation (ASF) under one
@@ -24,17 +24,35 @@ specific language governing permissions and limitations
under the License.
-->
-# Append Queue Table
+# Append Queue
## Definition
In this mode, you can regard append table as a queue separated by bucket. Every record in the same bucket is ordered strictly,
streaming read will transfer the record to down-stream exactly in the order of writing. To use this mode, you do not need
to config special configurations, all the data will go into one bucket as a queue. You can also define the `bucket` and
-`bucket-key` to enable larger parallelism and disperse data (see [Example]({{< ref "#example" >}})).
+`bucket-key` to enable larger parallelism and disperse data.
{{< img src="/img/for-queue.png">}}
+Example to create append queue table:
+
+{{< tabs "create-append-queue" >}}
+{{< tab "Flink" >}}
+
+```sql
+CREATE TABLE my_table (
+ product_id BIGINT,
+ price DOUBLE,
+ sales BIGINT
+) WITH (
+ 'bucket' = '8',
+ 'bucket-key' = 'product_id'
+);
+```
+{{< /tab >}}
+{{< /tabs >}}
+
## Compaction
By default, the sink node will automatically perform compaction to control the number of files. The following options
@@ -158,25 +176,3 @@ INSERT INTO paimon_table SELECT * FROM kakfa_table;
-- launch a bounded streaming job to read paimon_table
SELECT * FROM paimon_table /*+ OPTIONS('scan.bounded.watermark'='...') */;
```
-
-## Example
-
-The following is an example of creating the Append table and specifying the bucket key.
-
-{{< tabs "create-append-table" >}}
-
-{{< tab "Flink" >}}
-
-```sql
-CREATE TABLE my_table (
- product_id BIGINT,
- price DOUBLE,
- sales BIGINT
-) WITH (
- 'bucket' = '8',
- 'bucket-key' = 'product_id'
- );
-```
-{{< /tab >}}
-
-{{< /tabs >}}
diff --git a/docs/content/append-table/append-scalable-table.md b/docs/content/append-table/append-scalable-table.md
deleted file mode 100644
index f2b80c202..000000000
--- a/docs/content/append-table/append-scalable-table.md
+++ /dev/null
@@ -1,114 +0,0 @@
----
-title: "Append Scalable Table"
-weight: 2
-type: docs
-aliases:
-- /append-table/append-scalable-table.html
----
-<!--
-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.
--->
-
-# Append Scalable Table
-
-## Definition
-
-By defining `'bucket' = '-1'` in table properties, you can assign a special mode (we call it "unaware-bucket mode") to this
-table (see [Example]({{< ref "#example" >}})). In this mode, all the things are different. We don't have
-the concept of bucket anymore, and we don't guarantee the order of streaming read. We regard this table as a batch off-line table (
-although we can stream read and write still). All the records will go into one directory (for compatibility, we put them in bucket-0),
-and we do not maintain the order anymore. As we don't have the concept of bucket, we will not shuffle the input records by bucket anymore,
-which will speed up the inserting.
-
-Using this mode, you can replace your Hive table to lake table.
-
-{{< img src="/img/for-scalable.png">}}
-
-## Compaction
-
-In unaware-bucket mode, we don't do compaction in writer, instead, we use `Compact Coordinator` to scan the small files and submit compaction task
-to `Compact Worker`. By this, we can easily do compaction for one simple data directory in parallel. In streaming mode, if you run insert sql in flink,
-the topology will be like this:
-
-{{< img src="/img/unaware-bucket-topo.png">}}
-
-It will do its best to compact small files, but when a single small file in one partition remains long time
-and no new file added to the partition, the `Compact Coordinator` will remove it from memory to reduce memory usage.
-After you restart the job, it will scan the small files and add it to memory again. The options to control the compact
-behavior is exactly the same as [Append For Qeueue]({{< ref "#compaction" >}}). If you set `write-only` to true, the
-`Compact Coordinator` and `Compact Worker` will be removed in the topology.
-
-The auto compaction is only supported in Flink engine streaming mode. You can also start a compaction job in flink by flink action in paimon
-and disable all the other compaction by set `write-only`.
-
-## Sort Compact
-
-The data in a per-partition out of order will lead a slow select, compaction may slow down the inserting. It is a good choice for you to set
-`write-only` for inserting job, and after per-partition data done, trigger a partition `Sort Compact` action. See [Sort Compact]({{< ref "maintenance/dedicated-compaction#sort-compact" >}}).
-
-## Streaming Source
-
-Unaware-bucket mode append table supported streaming read and write, but no longer guarantee order anymore. You cannot regard it
-as a queue, instead, as a lake with storage bins. Every commit will generate a new record bin, we can read the
-increase by reading the new record bin, but records in one bin are flowing to anywhere they want, and we fetch them in any possible order.
-While in the `Append For Queue` mode, records are not stored in bins, but in record pipe. We can see the difference below.
-
-## Streaming Multiple Partitions Write
-
-Since the number of write tasks that Paimon-sink needs to handle is: the number of partitions to which the data is written * the number of buckets per partition.
-Therefore, we need to try to control the number of write tasks per paimon-sink task as much as possible,so that it is distributed within a reasonable range.
-If each sink-task handles too many write tasks, not only will it cause problems with too many small files, but it may also lead to out-of-memory errors.
-
-In addition, write failures introduce orphan files, which undoubtedly adds to the cost of maintaining paimon. We need to avoid this problem as much as possible.
-
-For flink-jobs with auto-merge enabled, we recommend trying to follow the following formula to adjust the parallelism of paimon-sink(This doesn't just apply to append-tables, it actually applies to most scenarios):
-```
-(N*B)/P < 100 (This value needs to be adjusted according to the actual situation)
-N(the number of partitions to which the data is written)
-B(bucket number)
-P(parallelism of paimon-sink)
-100 (This is an empirically derived threshold,For flink-jobs with auto-merge disabled, this value can be reduced.
-However, please note that you are only transferring part of the work to the user-compaction-job, you still have to deal with the problem in essence,
-the amount of work you have to deal with has not been reduced, and the user-compaction-job still needs to be adjusted according to the above formula.)
-```
-You can also set `write-buffer-spillable` to true, writer can spill the records to disk. This can reduce small
-files as much as possible.To use this option, you need to have a certain size of local disks for your flink cluster. This is especially important for those using flink on k8s.
-
-For append-table, You can set `write-buffer-for-append` option for append table. Setting this parameter to true, writer will cache
-the records use Segment Pool to avoid OOM.
-
-## Example
-
-The following is an example of creating the Append table and specifying the bucket key.
-
-{{< tabs "create-append-table-unaware-bucket" >}}
-
-{{< tab "Flink" >}}
-
-```sql
-CREATE TABLE my_table (
- product_id BIGINT,
- price DOUBLE,
- sales BIGINT
-) WITH (
- 'bucket' = '-1'
-);
-```
-{{< /tab >}}
-
-{{< /tabs >}}
diff --git a/docs/content/append-table/append-table.md b/docs/content/append-table/append-table.md
new file mode 100644
index 000000000..239a19646
--- /dev/null
+++ b/docs/content/append-table/append-table.md
@@ -0,0 +1,64 @@
+---
+title: "Append Table"
+weight: 1
+type: docs
+aliases:
+- /append-table/append-table.html
+---
+<!--
+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.
+-->
+
+# Append Table
+
+If a table does not have a primary key defined, it is an append table by default.
+
+You can only insert a complete record into the table in streaming. This type of table is suitable for use cases that
+do not require streaming updates (such as log data synchronization).
+
+{{< tabs "create-append-table" >}}
+{{< tab "Flink" >}}
+```sql
+CREATE TABLE my_table (
+ product_id BIGINT,
+ price DOUBLE,
+ sales BIGINT
+);
+```
+{{< /tab >}}
+{{< /tabs >}}
+
+## Data Distribution
+
+By default, append table has no bucket concept. It acts just like a Hive Table. The data files are placed under
+partitions where they can be reorganized and reordered to speed up queries.
+
+## Automatic small file merging
+
+In streaming writing job, without bucket definition, there is no compaction in writer, instead, will use
+`Compact Coordinator` to scan the small files and pass compaction task to `Compact Worker`. In streaming mode, if you
+run insert sql in flink, the topology will be like this:
+
+{{< img src="/img/unaware-bucket-topo.png">}}
+
+Do not worry about backpressure, compaction never backpressure.
+
+If you set `write-only` to true, the `Compact Coordinator` and `Compact Worker` will be removed in the topology.
+
+The auto compaction is only supported in Flink engine streaming mode. You can also start a compaction job in flink by
+flink action in paimon and disable all the other compaction by set `write-only`.
\ No newline at end of file
diff --git a/docs/content/append-table/overview.md b/docs/content/append-table/overview.md
deleted file mode 100644
index 5c0f4c560..000000000
--- a/docs/content/append-table/overview.md
+++ /dev/null
@@ -1,36 +0,0 @@
----
-title: "Overview"
-weight: 1
-type: docs
-aliases:
-- /append-table/overview.html
----
-<!--
-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.
--->
-
-# Overview
-
-If a table does not have a primary key defined, it is an append table by default.
-
-You can only insert a complete record into the table in streaming. This type of table is suitable for use cases that
-do not require streaming updates (such as log data synchronization).
-
-{{< hint info >}}
-We highly recommend using [Append Scalable Table]({{< ref "append-table/append-scalable-table" >}}). (bucket = -1).
-{{< /hint >}}
\ No newline at end of file
diff --git a/docs/content/learn-paimon/understand-files.md b/docs/content/learn-paimon/understand-files.md
index adbc614ba..00129c2c0 100644
--- a/docs/content/learn-paimon/understand-files.md
+++ b/docs/content/learn-paimon/understand-files.md
@@ -455,7 +455,7 @@ this means that there are at least 5 files in a bucket. If you want to reduce th
By default, Append also does automatic compaction to reduce the number of small files.
However, for Bucket's Append table, it will only compact the files within the Bucket for sequential
-purposes, which may keep more small files. See [Append Queue Table]({{< ref "append-table/append-queue-table" >}}).
+purposes, which may keep more small files. See [Append Queue]({{< ref "append-table/append-queue" >}}).
### Understand Full-Compaction
diff --git a/docs/content/maintenance/dedicated-compaction.md b/docs/content/maintenance/dedicated-compaction.md
index 41f0e6c91..f000849bf 100644
--- a/docs/content/maintenance/dedicated-compaction.md
+++ b/docs/content/maintenance/dedicated-compaction.md
@@ -229,7 +229,7 @@ For more usage of the compact_database action, see
## Sort Compact
If your table is configured with [dynamic bucket primary key table]({{< ref "primary-key-table/data-distribution#dynamic-bucket" >}})
-or [unaware bucket append table]({{< ref "append-table/append-scalable-table" >}}) ,
+or [append table]({{< ref "append-table/append-table" >}}) ,
you can trigger a compact with specified column sort to speed up queries.
```bash
diff --git a/docs/content/migration/migration-from-hive.md b/docs/content/migration/migration-from-hive.md
index 4f0383625..dd1132444 100644
--- a/docs/content/migration/migration-from-hive.md
+++ b/docs/content/migration/migration-from-hive.md
@@ -28,7 +28,7 @@ under the License.
Apache Hive supports ORC, Parquet file formats that could be migrated to Paimon.
When migrating data to a paimon table, the origin table will be permanently disappeared. So please back up your data if you
-still need the original table. The migrated table will be [unaware-bucket append table]({{< ref "append-table/append-scalable-table" >}}).
+still need the original table. The migrated table will be [append table]({{< ref "append-table/append-table" >}}).
Now, we can use paimon hive catalog with Migrate Table Procedure and Migrate File Procedure to totally migrate a table from hive to paimon.
At the same time, you can use paimon hive catalog with Migrate Database Procedure to fully synchronize all tables in the database to paimon.
diff --git a/docs/static/img/for-scalable.png b/docs/static/img/for-scalable.png
deleted file mode 100644
index ea5a015c2..000000000
Binary files a/docs/static/img/for-scalable.png and /dev/null differ
(paimon) 21/31: [core] Revert branchs in TagsTable (#3350)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 3881d6a8879726902c01336290d5e7ced5893265
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Mon May 20 18:21:15 2024 +0800
[core] Revert branchs in TagsTable (#3350)
---
.../org/apache/paimon/table/system/TagsTable.java | 35 ++++------------------
.../apache/paimon/table/system/TagsTableTest.java | 35 ++--------------------
2 files changed, 8 insertions(+), 62 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/system/TagsTable.java b/paimon-core/src/main/java/org/apache/paimon/table/system/TagsTable.java
index 5a179cfc1..0c1c4aa29 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/system/TagsTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/system/TagsTable.java
@@ -18,7 +18,6 @@
package org.apache.paimon.table.system;
-import org.apache.paimon.CoreOptions;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
@@ -26,11 +25,8 @@ import org.apache.paimon.data.Timestamp;
import org.apache.paimon.disk.IOManager;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.Path;
-import org.apache.paimon.options.Options;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.reader.RecordReader;
-import org.apache.paimon.table.FileStoreTable;
-import org.apache.paimon.table.FileStoreTableFactory;
import org.apache.paimon.table.ReadonlyTable;
import org.apache.paimon.table.Table;
import org.apache.paimon.table.source.InnerTableRead;
@@ -52,10 +48,8 @@ import org.apache.paimon.utils.TagManager;
import org.apache.paimon.shade.guava30.com.google.common.collect.Iterators;
-import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
-import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
@@ -80,10 +74,9 @@ public class TagsTable implements ReadonlyTable {
new DataField(2, "schema_id", new BigIntType(false)),
new DataField(3, "commit_time", new TimestampType(false, 3)),
new DataField(4, "record_count", new BigIntType(true)),
- new DataField(5, "branches", SerializationUtils.newStringType(true)),
- new DataField(6, "create_time", new TimestampType(true, 3)),
+ new DataField(5, "create_time", new TimestampType(true, 3)),
new DataField(
- 7, "time_retained", SerializationUtils.newStringType(true))));
+ 6, "time_retained", SerializationUtils.newStringType(true))));
private final FileIO fileIO;
private final Path location;
@@ -140,6 +133,7 @@ public class TagsTable implements ReadonlyTable {
}
private static class TagsSplit implements Split {
+
private static final long serialVersionUID = 1L;
private final long rowCount;
@@ -205,28 +199,14 @@ public class TagsTable implements ReadonlyTable {
throw new IllegalArgumentException("Unsupported split: " + split.getClass());
}
Path location = ((TagsSplit) split).location;
- Options options = new Options();
- options.set(CoreOptions.PATH, location.toUri().toString());
- FileStoreTable table = FileStoreTableFactory.create(fileIO, options);
- List<Pair<Tag, String>> tags = table.tagManager().tagObjects();
+ List<Pair<Tag, String>> tags = new TagManager(fileIO, location).tagObjects();
Map<String, Tag> nameToSnapshot = new LinkedHashMap<>();
for (Pair<Tag, String> tag : tags) {
nameToSnapshot.put(tag.getValue(), tag.getKey());
}
- Map<String, List<String>> tagBranches = new HashMap<>();
- table.branchManager()
- .branches()
- .forEach(
- branch ->
- tagBranches
- .computeIfAbsent(
- branch.getCreatedFromTag(),
- key -> new ArrayList<>())
- .add(branch.getBranchName()));
Iterator<InternalRow> rows =
- Iterators.transform(
- nameToSnapshot.entrySet().iterator(), tag -> toRow(tag, tagBranches));
+ Iterators.transform(nameToSnapshot.entrySet().iterator(), this::toRow);
if (projection != null) {
rows =
Iterators.transform(
@@ -235,17 +215,14 @@ public class TagsTable implements ReadonlyTable {
return new IteratorRecordReader<>(rows);
}
- private InternalRow toRow(
- Map.Entry<String, Tag> snapshot, Map<String, List<String>> tagBranches) {
+ private InternalRow toRow(Map.Entry<String, Tag> snapshot) {
Tag tag = snapshot.getValue();
- List<String> branches = tagBranches.get(snapshot.getKey());
return GenericRow.of(
BinaryString.fromString(snapshot.getKey()),
tag.id(),
tag.schemaId(),
Timestamp.fromLocalDateTime(DateTimeUtils.toLocalDateTime(tag.timeMillis())),
tag.totalRecordCount(),
- BinaryString.fromString(branches == null ? "[]" : branches.toString()),
Optional.ofNullable(tag.getTagCreateTime())
.map(Timestamp::fromLocalDateTime)
.orElse(null),
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/system/TagsTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/system/TagsTableTest.java
index a1b7770c5..99c38df01 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/system/TagsTableTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/system/TagsTableTest.java
@@ -26,7 +26,6 @@ import org.apache.paimon.data.Timestamp;
import org.apache.paimon.manifest.ManifestCommittable;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.table.FileStoreTable;
-import org.apache.paimon.table.Table;
import org.apache.paimon.table.TableTestBase;
import org.apache.paimon.table.sink.TableCommitImpl;
import org.apache.paimon.tag.Tag;
@@ -40,10 +39,7 @@ import org.junit.jupiter.api.Test;
import java.time.LocalDateTime;
import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
import java.util.List;
-import java.util.function.Function;
import static org.assertj.core.api.Assertions.assertThat;
@@ -87,38 +83,12 @@ class TagsTableTest extends TableTestBase {
@Test
void testTagsTable() throws Exception {
- List<InternalRow> expectRow =
- getExceptedResult(
- key -> {
- return new ArrayList<>();
- });
+ List<InternalRow> expectRow = getExceptedResult();
List<InternalRow> result = read(tagsTable);
assertThat(result).containsExactlyElementsOf(expectRow);
}
- @Test
- void testTagBranchesTable() throws Exception {
- Table table = catalog.getTable(identifier(tableName));
- table.createBranch("2023-07-17-branch1", "2023-07-17");
- table.createBranch("2023-07-18-branch1", "2023-07-18");
- table.createBranch("2023-07-18-branch2", "2023-07-18");
- List<InternalRow> expectRow =
- getExceptedResult(
- tag -> {
- if (tag.equals("2023-07-17")) {
- return Collections.singletonList("2023-07-17-branch1");
- } else if (tag.equals("2023-07-18")) {
- return Arrays.asList("2023-07-18-branch1", "2023-07-18-branch2");
- } else {
- return new ArrayList<>();
- }
- });
- List<InternalRow> result = read(tagsTable);
- assertThat(result).containsExactlyElementsOf(expectRow);
- }
-
- private List<InternalRow> getExceptedResult(
- Function<String, List<String>> tagBranchesFunction) {
+ private List<InternalRow> getExceptedResult() {
List<InternalRow> internalRows = new ArrayList<>();
for (Pair<Tag, String> snapshot : tagManager.tagObjects()) {
Tag tag = snapshot.getKey();
@@ -131,7 +101,6 @@ class TagsTableTest extends TableTestBase {
Timestamp.fromLocalDateTime(
DateTimeUtils.toLocalDateTime(tag.timeMillis())),
tag.totalRecordCount(),
- BinaryString.fromString(tagBranchesFunction.apply(tagName).toString()),
tag.getTagCreateTime() == null
? null
: Timestamp.fromLocalDateTime(tag.getTagCreateTime()),
(paimon) 23/31: [flink] Adjust the setMaxParallelism api call in QueryService (#3377)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 78a8509372a745b41fa460903cfdb16764c302b8
Author: Aitozi <ai...@apache.org>
AuthorDate: Wed May 22 17:15:51 2024 +0800
[flink] Adjust the setMaxParallelism api call in QueryService (#3377)
---
.../apache/paimon/flink/service/QueryFileMonitor.java | 5 +++++
.../org/apache/paimon/flink/service/QueryService.java | 19 +++++++++++--------
2 files changed, 16 insertions(+), 8 deletions(-)
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/service/QueryFileMonitor.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/service/QueryFileMonitor.java
index 4cd34ab26..43cf654e9 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/service/QueryFileMonitor.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/service/QueryFileMonitor.java
@@ -142,5 +142,10 @@ public class QueryFileMonitor extends RichSourceFunction<InternalRow> {
int bucket = row.getInt(2);
return ChannelComputer.select(partition, bucket, numChannels);
}
+
+ @Override
+ public String toString() {
+ return "FileMonitorChannelComputer{" + "numChannels=" + numChannels + '}';
+ }
}
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/service/QueryService.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/service/QueryService.java
index 8a4814d0a..2066a1513 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/service/QueryService.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/service/QueryService.java
@@ -29,6 +29,7 @@ import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.configuration.ExecutionOptions;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import static org.apache.paimon.flink.sink.FlinkStreamPartitioner.partition;
@@ -55,13 +56,15 @@ public class QueryService {
stream = partition(stream, QueryFileMonitor.createChannelComputer(), parallelism);
QueryExecutorOperator executorOperator = new QueryExecutorOperator(table);
- stream.transform(
- "Executor",
- InternalTypeInfo.fromRowType(QueryExecutorOperator.outputType()),
- executorOperator)
- .setParallelism(parallelism)
- .addSink(new QueryAddressRegister(table))
- .setParallelism(1)
- .setMaxParallelism(1);
+ DataStreamSink<?> sink =
+ stream.transform(
+ "Executor",
+ InternalTypeInfo.fromRowType(QueryExecutorOperator.outputType()),
+ executorOperator)
+ .setParallelism(parallelism)
+ .addSink(new QueryAddressRegister(table))
+ .setParallelism(1);
+
+ sink.getTransformation().setMaxParallelism(1);
}
}
(paimon) 18/31: [core] Incremental-between tags should deduplicate records (#3338)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit de32dae41a165984d6fe121809e991ab75881ae5
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Thu May 16 17:03:12 2024 +0800
[core] Incremental-between tags should deduplicate records (#3338)
---
.../org/apache/paimon/mergetree/MergeSorter.java | 4 +
.../paimon/operation/MergeFileSplitRead.java | 109 ++++++++-------------
.../apache/paimon/operation/RawFileSplitRead.java | 12 +++
.../org/apache/paimon/operation/SplitRead.java | 46 ++++++++-
.../paimon/table/source/KeyValueTableRead.java | 79 +++++++--------
.../IncrementalChangelogReadProvider.java | 96 ++++++++++++++++++
.../splitread/IncrementalDiffReadProvider.java | 65 ++++++++++++
.../splitread/IncrementalDiffSplitRead.java} | 95 ++++++++++++++++--
.../splitread/MergeFileSplitReadProvider.java | 68 +++++++++++++
.../source/splitread/RawFileSplitReadProvider.java | 60 ++++++++++++
.../source/splitread/SplitReadProvider.java} | 21 ++--
.../paimon/operation/MergeFileSplitReadTest.java | 2 +-
.../apache/paimon/table/IncrementalTableTest.java | 18 +++-
13 files changed, 541 insertions(+), 134 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/mergetree/MergeSorter.java b/paimon-core/src/main/java/org/apache/paimon/mergetree/MergeSorter.java
index 0f54b40b6..420613899 100644
--- a/paimon-core/src/main/java/org/apache/paimon/mergetree/MergeSorter.java
+++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/MergeSorter.java
@@ -95,6 +95,10 @@ public class MergeSorter {
return memoryPool;
}
+ public RowType valueType() {
+ return valueType;
+ }
+
public void setIOManager(IOManager ioManager) {
this.ioManager = ioManager;
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/MergeFileSplitRead.java b/paimon-core/src/main/java/org/apache/paimon/operation/MergeFileSplitRead.java
index d457c9093..8002b62f0 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/MergeFileSplitRead.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/MergeFileSplitRead.java
@@ -116,7 +116,16 @@ public class MergeFileSplitRead implements SplitRead<KeyValue> {
return this;
}
- public MergeFileSplitRead withValueProjection(@Nullable int[][] projectedFields) {
+ public Comparator<InternalRow> keyComparator() {
+ return keyComparator;
+ }
+
+ public MergeSorter mergeSorter() {
+ return mergeSorter;
+ }
+
+ @Override
+ public MergeFileSplitRead withProjection(@Nullable int[][] projectedFields) {
if (projectedFields == null) {
return this;
}
@@ -160,11 +169,13 @@ public class MergeFileSplitRead implements SplitRead<KeyValue> {
return this;
}
+ @Override
public MergeFileSplitRead withIOManager(IOManager ioManager) {
this.mergeSorter.setIOManager(ioManager);
return this;
}
+ @Override
public MergeFileSplitRead forceKeepDelete() {
this.forceKeepDelete = true;
return this;
@@ -211,75 +222,28 @@ public class MergeFileSplitRead implements SplitRead<KeyValue> {
@Override
public RecordReader<KeyValue> createReader(DataSplit split) throws IOException {
- RecordReader<KeyValue> reader = createReaderWithoutOuterProjection(split);
- if (outerProjection != null) {
- ProjectedRow projectedRow = ProjectedRow.from(outerProjection);
- reader = reader.transform(kv -> kv.replaceValue(projectedRow.replaceRow(kv.value())));
+ if (!split.beforeFiles().isEmpty()) {
+ throw new IllegalArgumentException("This read cannot accept split with before files.");
}
- return reader;
- }
- private RecordReader<KeyValue> createReaderWithoutOuterProjection(DataSplit split)
- throws IOException {
- if (split.beforeFiles().isEmpty()) {
- if (split.isStreaming() || split.convertToRawFiles().isPresent()) {
- return noMergeRead(
- split.partition(),
- split.bucket(),
- split.dataFiles(),
- split.deletionFiles().orElse(null),
- split.isStreaming());
- } else {
- return projectKey(
- mergeRead(
- split.partition(),
- split.bucket(),
- split.dataFiles(),
- null,
- forceKeepDelete));
- }
- } else if (split.isStreaming()) {
- // streaming concat read
- return ConcatRecordReader.create(
- () ->
- new ReverseReader(
- noMergeRead(
- split.partition(),
- split.bucket(),
- split.beforeFiles(),
- split.beforeDeletionFiles().orElse(null),
- true)),
- () ->
- noMergeRead(
- split.partition(),
- split.bucket(),
- split.dataFiles(),
- split.deletionFiles().orElse(null),
- true));
+ if (split.isStreaming()) {
+ return createNoMergeReader(
+ split.partition(),
+ split.bucket(),
+ split.dataFiles(),
+ split.deletionFiles().orElse(null),
+ split.isStreaming());
} else {
- // batch diff read
- return projectKey(
- DiffReader.readDiff(
- mergeRead(
- split.partition(),
- split.bucket(),
- split.beforeFiles(),
- split.beforeDeletionFiles().orElse(null),
- false),
- mergeRead(
- split.partition(),
- split.bucket(),
- split.dataFiles(),
- split.deletionFiles().orElse(null),
- false),
- keyComparator,
- createUdsComparator(),
- mergeSorter,
- forceKeepDelete));
+ return createMergeReader(
+ split.partition(),
+ split.bucket(),
+ split.dataFiles(),
+ split.deletionFiles().orElse(null),
+ forceKeepDelete);
}
}
- private RecordReader<KeyValue> mergeRead(
+ public RecordReader<KeyValue> createMergeReader(
BinaryRow partition,
int bucket,
List<DataFileMeta> files,
@@ -316,10 +280,10 @@ public class MergeFileSplitRead implements SplitRead<KeyValue> {
reader = new DropDeleteReader(reader);
}
- return reader;
+ return projectOuter(projectKey(reader));
}
- private RecordReader<KeyValue> noMergeRead(
+ public RecordReader<KeyValue> createNoMergeReader(
BinaryRow partition,
int bucket,
List<DataFileMeta> files,
@@ -344,7 +308,8 @@ public class MergeFileSplitRead implements SplitRead<KeyValue> {
file.schemaId(), fileName, file.fileSize(), file.level());
});
}
- return ConcatRecordReader.create(suppliers);
+
+ return projectOuter(ConcatRecordReader.create(suppliers));
}
private Optional<String> changelogFile(DataFileMeta fileMeta) {
@@ -365,8 +330,16 @@ public class MergeFileSplitRead implements SplitRead<KeyValue> {
return reader.transform(kv -> kv.replaceKey(projectedRow.replaceRow(kv.key())));
}
+ private RecordReader<KeyValue> projectOuter(RecordReader<KeyValue> reader) {
+ if (outerProjection != null) {
+ ProjectedRow projectedRow = ProjectedRow.from(outerProjection);
+ reader = reader.transform(kv -> kv.replaceValue(projectedRow.replaceRow(kv.value())));
+ }
+ return reader;
+ }
+
@Nullable
- private UserDefinedSeqComparator createUdsComparator() {
+ public UserDefinedSeqComparator createUdsComparator() {
return UserDefinedSeqComparator.create(
readerFactoryBuilder.projectedValueType(), sequenceFields);
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/RawFileSplitRead.java b/paimon-core/src/main/java/org/apache/paimon/operation/RawFileSplitRead.java
index fb68823ba..083db39cb 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/RawFileSplitRead.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/RawFileSplitRead.java
@@ -23,6 +23,7 @@ import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.deletionvectors.ApplyDeletionVectorReader;
import org.apache.paimon.deletionvectors.DeletionVector;
+import org.apache.paimon.disk.IOManager;
import org.apache.paimon.format.FileFormatDiscover;
import org.apache.paimon.format.FormatKey;
import org.apache.paimon.format.FormatReaderContext;
@@ -98,6 +99,17 @@ public class RawFileSplitRead implements SplitRead<InternalRow> {
this.projection = Projection.range(0, rowType.getFieldCount()).toNestedIndexes();
}
+ @Override
+ public SplitRead<InternalRow> forceKeepDelete() {
+ return this;
+ }
+
+ @Override
+ public SplitRead<InternalRow> withIOManager(@Nullable IOManager ioManager) {
+ return this;
+ }
+
+ @Override
public RawFileSplitRead withProjection(int[][] projectedFields) {
if (projectedFields != null) {
projection = projectedFields;
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/SplitRead.java b/paimon-core/src/main/java/org/apache/paimon/operation/SplitRead.java
index a5b05e8f4..c17c0f6b3 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/SplitRead.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/SplitRead.java
@@ -18,9 +18,13 @@
package org.apache.paimon.operation;
+import org.apache.paimon.disk.IOManager;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.utils.IOFunction;
+
+import javax.annotation.Nullable;
import java.io.IOException;
@@ -31,8 +35,48 @@ import java.io.IOException;
*/
public interface SplitRead<T> {
- SplitRead<T> withFilter(Predicate predicate);
+ SplitRead<T> forceKeepDelete();
+
+ SplitRead<T> withIOManager(@Nullable IOManager ioManager);
+
+ SplitRead<T> withProjection(@Nullable int[][] projectedFields);
+
+ SplitRead<T> withFilter(@Nullable Predicate predicate);
/** Create a {@link RecordReader} from split. */
RecordReader<T> createReader(DataSplit split) throws IOException;
+
+ static <L, R> SplitRead<R> convert(
+ SplitRead<L> read, IOFunction<DataSplit, RecordReader<R>> convertedFactory) {
+ return new SplitRead<R>() {
+ @Override
+ public SplitRead<R> forceKeepDelete() {
+ read.forceKeepDelete();
+ return this;
+ }
+
+ @Override
+ public SplitRead<R> withIOManager(@Nullable IOManager ioManager) {
+ read.withIOManager(ioManager);
+ return this;
+ }
+
+ @Override
+ public SplitRead<R> withProjection(@Nullable int[][] projectedFields) {
+ read.withProjection(projectedFields);
+ return this;
+ }
+
+ @Override
+ public SplitRead<R> withFilter(@Nullable Predicate predicate) {
+ read.withFilter(predicate);
+ return this;
+ }
+
+ @Override
+ public RecordReader<R> createReader(DataSplit split) throws IOException {
+ return convertedFactory.apply(split);
+ }
+ };
+ }
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/KeyValueTableRead.java b/paimon-core/src/main/java/org/apache/paimon/table/source/KeyValueTableRead.java
index f3f66c4aa..c674e4792 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/source/KeyValueTableRead.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/KeyValueTableRead.java
@@ -23,14 +23,22 @@ import org.apache.paimon.data.InternalRow;
import org.apache.paimon.disk.IOManager;
import org.apache.paimon.operation.MergeFileSplitRead;
import org.apache.paimon.operation.RawFileSplitRead;
+import org.apache.paimon.operation.SplitRead;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.schema.TableSchema;
-import org.apache.paimon.utils.LazyField;
+import org.apache.paimon.table.source.splitread.IncrementalChangelogReadProvider;
+import org.apache.paimon.table.source.splitread.IncrementalDiffReadProvider;
+import org.apache.paimon.table.source.splitread.MergeFileSplitReadProvider;
+import org.apache.paimon.table.source.splitread.RawFileSplitReadProvider;
+import org.apache.paimon.table.source.splitread.SplitReadProvider;
import javax.annotation.Nullable;
import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
import java.util.function.Supplier;
/**
@@ -38,8 +46,7 @@ import java.util.function.Supplier;
*/
public final class KeyValueTableRead extends AbstractDataTableRead<KeyValue> {
- private final LazyField<MergeFileSplitRead> mergeRead;
- private final LazyField<RawFileSplitRead> batchRawRead;
+ private final List<SplitReadProvider> readProviders;
private int[][] projection = null;
private boolean forceKeepDelete = false;
@@ -51,65 +58,54 @@ public final class KeyValueTableRead extends AbstractDataTableRead<KeyValue> {
Supplier<RawFileSplitRead> batchRawReadSupplier,
TableSchema schema) {
super(schema);
- this.mergeRead = new LazyField<>(() -> createMergeRead(mergeReadSupplier));
- this.batchRawRead = new LazyField<>(() -> createBatchRawRead(batchRawReadSupplier));
+ this.readProviders =
+ Arrays.asList(
+ new RawFileSplitReadProvider(batchRawReadSupplier, this::assignValues),
+ new MergeFileSplitReadProvider(mergeReadSupplier, this::assignValues),
+ new IncrementalChangelogReadProvider(mergeReadSupplier, this::assignValues),
+ new IncrementalDiffReadProvider(mergeReadSupplier, this::assignValues));
}
- private MergeFileSplitRead createMergeRead(Supplier<MergeFileSplitRead> readSupplier) {
- MergeFileSplitRead read =
- readSupplier
- .get()
- .withKeyProjection(new int[0][])
- .withValueProjection(projection)
- .withFilter(predicate)
- .withIOManager(ioManager);
- if (forceKeepDelete) {
- read = read.forceKeepDelete();
+ private List<SplitRead<InternalRow>> initialized() {
+ List<SplitRead<InternalRow>> readers = new ArrayList<>();
+ for (SplitReadProvider readProvider : readProviders) {
+ if (readProvider.initialized()) {
+ readers.add(readProvider.getOrCreate());
+ }
}
- return read;
+ return readers;
}
- private RawFileSplitRead createBatchRawRead(Supplier<RawFileSplitRead> readSupplier) {
- return readSupplier.get().withProjection(projection).withFilter(predicate);
+ private void assignValues(SplitRead<InternalRow> read) {
+ if (forceKeepDelete) {
+ read = read.forceKeepDelete();
+ }
+ read.withProjection(projection).withFilter(predicate).withIOManager(ioManager);
}
@Override
public void projection(int[][] projection) {
- if (mergeRead.initialized()) {
- mergeRead.get().withValueProjection(projection);
- }
- if (batchRawRead.initialized()) {
- batchRawRead.get().withProjection(projection);
- }
+ initialized().forEach(r -> r.withProjection(projection));
this.projection = projection;
}
@Override
public InnerTableRead forceKeepDelete() {
- if (mergeRead.initialized()) {
- mergeRead.get().forceKeepDelete();
- }
+ initialized().forEach(SplitRead::forceKeepDelete);
this.forceKeepDelete = true;
return this;
}
@Override
protected InnerTableRead innerWithFilter(Predicate predicate) {
- if (mergeRead.initialized()) {
- mergeRead.get().withFilter(predicate);
- }
- if (batchRawRead.initialized()) {
- batchRawRead.get().withFilter(predicate);
- }
+ initialized().forEach(r -> r.withFilter(predicate));
this.predicate = predicate;
return this;
}
@Override
public TableRead withIOManager(IOManager ioManager) {
- if (mergeRead.initialized()) {
- mergeRead.get().withIOManager(ioManager);
- }
+ initialized().forEach(r -> r.withIOManager(ioManager));
this.ioManager = ioManager;
return this;
}
@@ -117,11 +113,16 @@ public final class KeyValueTableRead extends AbstractDataTableRead<KeyValue> {
@Override
public RecordReader<InternalRow> reader(Split split) throws IOException {
DataSplit dataSplit = (DataSplit) split;
- if (!forceKeepDelete && !dataSplit.isStreaming() && split.convertToRawFiles().isPresent()) {
- return batchRawRead.get().createReader(dataSplit);
+ for (SplitReadProvider readProvider : readProviders) {
+ if (readProvider.match(dataSplit, forceKeepDelete)) {
+ return readProvider.getOrCreate().createReader(dataSplit);
+ }
}
- RecordReader<KeyValue> reader = mergeRead.get().createReader(dataSplit);
+ throw new RuntimeException("Should not happen.");
+ }
+
+ public static RecordReader<InternalRow> unwrap(RecordReader<KeyValue> reader) {
return new RecordReader<InternalRow>() {
@Nullable
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/splitread/IncrementalChangelogReadProvider.java b/paimon-core/src/main/java/org/apache/paimon/table/source/splitread/IncrementalChangelogReadProvider.java
new file mode 100644
index 000000000..bec95979d
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/splitread/IncrementalChangelogReadProvider.java
@@ -0,0 +1,96 @@
+/*
+ * 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.table.source.splitread;
+
+import org.apache.paimon.KeyValue;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.mergetree.compact.ConcatRecordReader;
+import org.apache.paimon.operation.MergeFileSplitRead;
+import org.apache.paimon.operation.ReverseReader;
+import org.apache.paimon.operation.SplitRead;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.utils.IOFunction;
+import org.apache.paimon.utils.LazyField;
+
+import java.util.function.Consumer;
+import java.util.function.Supplier;
+
+import static org.apache.paimon.table.source.KeyValueTableRead.unwrap;
+
+/** A {@link SplitReadProvider} to incremental changelog read. */
+public class IncrementalChangelogReadProvider implements SplitReadProvider {
+
+ private final LazyField<SplitRead<InternalRow>> splitRead;
+
+ public IncrementalChangelogReadProvider(
+ Supplier<MergeFileSplitRead> supplier,
+ Consumer<SplitRead<InternalRow>> valuesAssigner) {
+ this.splitRead =
+ new LazyField<>(
+ () -> {
+ SplitRead<InternalRow> read = create(supplier);
+ valuesAssigner.accept(read);
+ return read;
+ });
+ }
+
+ private SplitRead<InternalRow> create(Supplier<MergeFileSplitRead> supplier) {
+ final MergeFileSplitRead read = supplier.get().withKeyProjection(new int[0][]);
+ IOFunction<DataSplit, RecordReader<InternalRow>> convertedFactory =
+ split -> {
+ RecordReader<KeyValue> reader =
+ ConcatRecordReader.create(
+ () ->
+ new ReverseReader(
+ read.createNoMergeReader(
+ split.partition(),
+ split.bucket(),
+ split.beforeFiles(),
+ split.beforeDeletionFiles()
+ .orElse(null),
+ true)),
+ () ->
+ read.createNoMergeReader(
+ split.partition(),
+ split.bucket(),
+ split.dataFiles(),
+ split.deletionFiles().orElse(null),
+ true));
+ return unwrap(reader);
+ };
+
+ return SplitRead.convert(read, convertedFactory);
+ }
+
+ @Override
+ public boolean match(DataSplit split, boolean forceKeepDelete) {
+ return !split.beforeFiles().isEmpty() && split.isStreaming();
+ }
+
+ @Override
+ public boolean initialized() {
+ return splitRead.initialized();
+ }
+
+ @Override
+ public SplitRead<InternalRow> getOrCreate() {
+ return splitRead.get();
+ }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/splitread/IncrementalDiffReadProvider.java b/paimon-core/src/main/java/org/apache/paimon/table/source/splitread/IncrementalDiffReadProvider.java
new file mode 100644
index 000000000..a335a7c03
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/splitread/IncrementalDiffReadProvider.java
@@ -0,0 +1,65 @@
+/*
+ * 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.table.source.splitread;
+
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.operation.MergeFileSplitRead;
+import org.apache.paimon.operation.SplitRead;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.utils.LazyField;
+
+import java.util.function.Consumer;
+import java.util.function.Supplier;
+
+/** A {@link SplitReadProvider} to batch incremental diff read. */
+public class IncrementalDiffReadProvider implements SplitReadProvider {
+
+ private final LazyField<SplitRead<InternalRow>> splitRead;
+
+ public IncrementalDiffReadProvider(
+ Supplier<MergeFileSplitRead> supplier,
+ Consumer<SplitRead<InternalRow>> valuesAssigner) {
+ this.splitRead =
+ new LazyField<>(
+ () -> {
+ SplitRead<InternalRow> read = create(supplier);
+ valuesAssigner.accept(read);
+ return read;
+ });
+ }
+
+ private SplitRead<InternalRow> create(Supplier<MergeFileSplitRead> supplier) {
+ return new IncrementalDiffSplitRead(supplier.get());
+ }
+
+ @Override
+ public boolean match(DataSplit split, boolean forceKeepDelete) {
+ return !split.beforeFiles().isEmpty() && !split.isStreaming();
+ }
+
+ @Override
+ public boolean initialized() {
+ return splitRead.initialized();
+ }
+
+ @Override
+ public SplitRead<InternalRow> getOrCreate() {
+ return splitRead.get();
+ }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/DiffReader.java b/paimon-core/src/main/java/org/apache/paimon/table/source/splitread/IncrementalDiffSplitRead.java
similarity index 55%
rename from paimon-core/src/main/java/org/apache/paimon/operation/DiffReader.java
rename to paimon-core/src/main/java/org/apache/paimon/table/source/splitread/IncrementalDiffSplitRead.java
index bc5153600..0519da9fd 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/DiffReader.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/splitread/IncrementalDiffSplitRead.java
@@ -16,15 +16,24 @@
* limitations under the License.
*/
-package org.apache.paimon.operation;
+package org.apache.paimon.table.source.splitread;
import org.apache.paimon.KeyValue;
import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.serializer.InternalRowSerializer;
+import org.apache.paimon.data.serializer.InternalSerializers;
+import org.apache.paimon.disk.IOManager;
import org.apache.paimon.mergetree.MergeSorter;
import org.apache.paimon.mergetree.compact.MergeFunctionWrapper;
+import org.apache.paimon.operation.MergeFileSplitRead;
+import org.apache.paimon.operation.SplitRead;
+import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.KeyValueTableRead;
import org.apache.paimon.types.RowKind;
import org.apache.paimon.utils.FieldsComparator;
+import org.apache.paimon.utils.ProjectedRow;
import javax.annotation.Nullable;
@@ -34,13 +43,73 @@ import java.util.Arrays;
import java.util.Comparator;
import java.util.List;
-/** A {@link RecordReader} util to read diff between before reader and after reader. */
-public class DiffReader {
+/** A {@link SplitRead} for batch incremental diff. */
+public class IncrementalDiffSplitRead implements SplitRead<InternalRow> {
private static final int BEFORE_LEVEL = Integer.MIN_VALUE;
private static final int AFTER_LEVEL = Integer.MAX_VALUE;
- public static RecordReader<KeyValue> readDiff(
+ private final MergeFileSplitRead mergeRead;
+
+ private boolean forceKeepDelete = false;
+ @Nullable private int[][] projectedFields;
+
+ public IncrementalDiffSplitRead(MergeFileSplitRead mergeRead) {
+ this.mergeRead = mergeRead;
+ }
+
+ @Override
+ public SplitRead<InternalRow> forceKeepDelete() {
+ this.forceKeepDelete = true;
+ return this;
+ }
+
+ @Override
+ public SplitRead<InternalRow> withIOManager(@Nullable IOManager ioManager) {
+ mergeRead.withIOManager(ioManager);
+ return this;
+ }
+
+ @Override
+ public SplitRead<InternalRow> withProjection(@Nullable int[][] projectedFields) {
+ this.projectedFields = projectedFields;
+ return this;
+ }
+
+ @Override
+ public SplitRead<InternalRow> withFilter(@Nullable Predicate predicate) {
+ mergeRead.withFilter(predicate);
+ return this;
+ }
+
+ @Override
+ public RecordReader<InternalRow> createReader(DataSplit split) throws IOException {
+ RecordReader<KeyValue> reader =
+ readDiff(
+ mergeRead.createMergeReader(
+ split.partition(),
+ split.bucket(),
+ split.beforeFiles(),
+ split.beforeDeletionFiles().orElse(null),
+ false),
+ mergeRead.createMergeReader(
+ split.partition(),
+ split.bucket(),
+ split.dataFiles(),
+ split.deletionFiles().orElse(null),
+ false),
+ mergeRead.keyComparator(),
+ mergeRead.createUdsComparator(),
+ mergeRead.mergeSorter(),
+ forceKeepDelete);
+ if (projectedFields != null) {
+ ProjectedRow projectedRow = ProjectedRow.from(projectedFields);
+ reader = reader.transform(kv -> kv.replaceValue(projectedRow.replaceRow(kv.value())));
+ }
+ return KeyValueTableRead.unwrap(reader);
+ }
+
+ private static RecordReader<KeyValue> readDiff(
RecordReader<KeyValue> beforeReader,
RecordReader<KeyValue> afterReader,
Comparator<InternalRow> keyComparator,
@@ -54,7 +123,7 @@ public class DiffReader {
() -> wrapLevelToReader(afterReader, AFTER_LEVEL)),
keyComparator,
userDefinedSeqComparator,
- new DiffMerger(keepDelete));
+ new DiffMerger(keepDelete, InternalSerializers.create(sorter.valueType())));
}
private static RecordReader<KeyValue> wrapLevelToReader(
@@ -96,11 +165,15 @@ public class DiffReader {
private static class DiffMerger implements MergeFunctionWrapper<KeyValue> {
private final boolean keepDelete;
+ private final InternalRowSerializer serializer1;
+ private final InternalRowSerializer serializer2;
private final List<KeyValue> kvs = new ArrayList<>();
- public DiffMerger(boolean keepDelete) {
+ public DiffMerger(boolean keepDelete, InternalRowSerializer serializer) {
this.keepDelete = keepDelete;
+ this.serializer1 = serializer;
+ this.serializer2 = serializer.duplicate();
}
@Override
@@ -128,7 +201,9 @@ public class DiffReader {
} else if (kvs.size() == 2) {
KeyValue latest = kvs.get(1);
if (latest.level() == AFTER_LEVEL) {
- return latest;
+ if (!valueEquals()) {
+ return latest;
+ }
}
} else {
throw new IllegalArgumentException("Illegal kv number: " + kvs.size());
@@ -136,5 +211,11 @@ public class DiffReader {
return null;
}
+
+ private boolean valueEquals() {
+ return serializer1
+ .toBinaryRow(kvs.get(0).value())
+ .equals(serializer2.toBinaryRow(kvs.get(1).value()));
+ }
}
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/splitread/MergeFileSplitReadProvider.java b/paimon-core/src/main/java/org/apache/paimon/table/source/splitread/MergeFileSplitReadProvider.java
new file mode 100644
index 000000000..abed0f33c
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/splitread/MergeFileSplitReadProvider.java
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.table.source.splitread;
+
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.operation.MergeFileSplitRead;
+import org.apache.paimon.operation.SplitRead;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.utils.LazyField;
+
+import java.util.function.Consumer;
+import java.util.function.Supplier;
+
+import static org.apache.paimon.table.source.KeyValueTableRead.unwrap;
+
+/** A {@link SplitReadProvider} to merge files. */
+public class MergeFileSplitReadProvider implements SplitReadProvider {
+
+ private final LazyField<SplitRead<InternalRow>> splitRead;
+
+ public MergeFileSplitReadProvider(
+ Supplier<MergeFileSplitRead> supplier,
+ Consumer<SplitRead<InternalRow>> valuesAssigner) {
+ this.splitRead =
+ new LazyField<>(
+ () -> {
+ SplitRead<InternalRow> read = create(supplier);
+ valuesAssigner.accept(read);
+ return read;
+ });
+ }
+
+ private SplitRead<InternalRow> create(Supplier<MergeFileSplitRead> supplier) {
+ final MergeFileSplitRead read = supplier.get().withKeyProjection(new int[0][]);
+ return SplitRead.convert(read, split -> unwrap(read.createReader(split)));
+ }
+
+ @Override
+ public boolean match(DataSplit split, boolean forceKeepDelete) {
+ return split.beforeFiles().isEmpty();
+ }
+
+ @Override
+ public boolean initialized() {
+ return splitRead.initialized();
+ }
+
+ @Override
+ public SplitRead<InternalRow> getOrCreate() {
+ return splitRead.get();
+ }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/splitread/RawFileSplitReadProvider.java b/paimon-core/src/main/java/org/apache/paimon/table/source/splitread/RawFileSplitReadProvider.java
new file mode 100644
index 000000000..9959ee555
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/splitread/RawFileSplitReadProvider.java
@@ -0,0 +1,60 @@
+/*
+ * 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.table.source.splitread;
+
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.operation.RawFileSplitRead;
+import org.apache.paimon.operation.SplitRead;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.utils.LazyField;
+
+import java.util.function.Consumer;
+import java.util.function.Supplier;
+
+/** A {@link SplitReadProvider} to create {@link RawFileSplitRead}. */
+public class RawFileSplitReadProvider implements SplitReadProvider {
+
+ private final LazyField<RawFileSplitRead> splitRead;
+
+ public RawFileSplitReadProvider(
+ Supplier<RawFileSplitRead> supplier, Consumer<SplitRead<InternalRow>> valuesAssigner) {
+ this.splitRead =
+ new LazyField<>(
+ () -> {
+ RawFileSplitRead read = supplier.get();
+ valuesAssigner.accept(read);
+ return read;
+ });
+ }
+
+ @Override
+ public boolean match(DataSplit split, boolean forceKeepDelete) {
+ return !forceKeepDelete && !split.isStreaming() && split.rawConvertible();
+ }
+
+ @Override
+ public boolean initialized() {
+ return splitRead.initialized();
+ }
+
+ @Override
+ public SplitRead<InternalRow> getOrCreate() {
+ return splitRead.get();
+ }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/SplitRead.java b/paimon-core/src/main/java/org/apache/paimon/table/source/splitread/SplitReadProvider.java
similarity index 64%
copy from paimon-core/src/main/java/org/apache/paimon/operation/SplitRead.java
copy to paimon-core/src/main/java/org/apache/paimon/table/source/splitread/SplitReadProvider.java
index a5b05e8f4..2aaefb322 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/SplitRead.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/source/splitread/SplitReadProvider.java
@@ -16,23 +16,18 @@
* limitations under the License.
*/
-package org.apache.paimon.operation;
+package org.apache.paimon.table.source.splitread;
-import org.apache.paimon.predicate.Predicate;
-import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.operation.SplitRead;
import org.apache.paimon.table.source.DataSplit;
-import java.io.IOException;
+/** Provider to create {@link SplitRead}. */
+public interface SplitReadProvider {
-/**
- * Read operation which provides {@link RecordReader} creation.
- *
- * @param <T> type of record to read.
- */
-public interface SplitRead<T> {
+ boolean match(DataSplit split, boolean forceKeepDelete);
- SplitRead<T> withFilter(Predicate predicate);
+ boolean initialized();
- /** Create a {@link RecordReader} from split. */
- RecordReader<T> createReader(DataSplit split) throws IOException;
+ SplitRead<InternalRow> getOrCreate();
}
diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/MergeFileSplitReadTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/MergeFileSplitReadTest.java
index 5652fcd43..1794e8aea 100644
--- a/paimon-core/src/test/java/org/apache/paimon/operation/MergeFileSplitReadTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/operation/MergeFileSplitReadTest.java
@@ -231,7 +231,7 @@ public class MergeFileSplitReadTest {
read.withKeyProjection(keyProjection);
}
if (valueProjection != null) {
- read.withValueProjection(valueProjection);
+ read.withProjection(valueProjection);
}
List<KeyValue> result = new ArrayList<>();
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/IncrementalTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/IncrementalTableTest.java
index f4dac9d11..b4b905d36 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/IncrementalTableTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/IncrementalTableTest.java
@@ -225,6 +225,8 @@ public class IncrementalTableTest extends TableTestBase {
GenericRow.of(1, 1, 1),
GenericRow.of(1, 2, 1),
GenericRow.of(1, 3, 1),
+ GenericRow.of(1, 4, 1),
+ GenericRow.of(1, 5, 1),
GenericRow.of(2, 1, 1));
// snapshot 2: append
@@ -235,7 +237,7 @@ public class IncrementalTableTest extends TableTestBase {
// UPDATE
GenericRow.of(1, 2, 2),
// NEW
- GenericRow.of(1, 4, 1));
+ GenericRow.of(1, 6, 1));
// snapshot 3: compact
compact(table, row(1), 0);
@@ -247,24 +249,30 @@ public class IncrementalTableTest extends TableTestBase {
// read tag1 tag2
List<InternalRow> result = read(table, Pair.of(INCREMENTAL_BETWEEN, "TAG1,TAG2"));
assertThat(result)
- .containsExactlyInAnyOrder(GenericRow.of(1, 2, 2), GenericRow.of(1, 4, 1));
+ .containsExactlyInAnyOrder(GenericRow.of(1, 2, 2), GenericRow.of(1, 6, 1));
result = read(auditLog, Pair.of(INCREMENTAL_BETWEEN, "TAG1,TAG2"));
assertThat(result)
.containsExactlyInAnyOrder(
GenericRow.of(fromString("-D"), 1, 1, 1),
GenericRow.of(fromString("+I"), 1, 2, 2),
- GenericRow.of(fromString("+I"), 1, 4, 1));
+ GenericRow.of(fromString("+I"), 1, 6, 1));
// read tag1 tag3
result = read(table, Pair.of(INCREMENTAL_BETWEEN, "TAG1,TAG3"));
assertThat(result)
- .containsExactlyInAnyOrder(GenericRow.of(1, 2, 2), GenericRow.of(1, 4, 1));
+ .containsExactlyInAnyOrder(GenericRow.of(1, 2, 2), GenericRow.of(1, 6, 1));
+
+ // read tag1 tag3 auditLog
result = read(auditLog, Pair.of(INCREMENTAL_BETWEEN, "TAG1,TAG3"));
assertThat(result)
.containsExactlyInAnyOrder(
GenericRow.of(fromString("-D"), 1, 1, 1),
GenericRow.of(fromString("+I"), 1, 2, 2),
- GenericRow.of(fromString("+I"), 1, 4, 1));
+ GenericRow.of(fromString("+I"), 1, 6, 1));
+
+ // read tag1 tag3 projection
+ result = read(table, new int[][] {{1}}, Pair.of(INCREMENTAL_BETWEEN, "TAG1,TAG3"));
+ assertThat(result).containsExactlyInAnyOrder(GenericRow.of(2), GenericRow.of(6));
assertThatThrownBy(() -> read(table, Pair.of(INCREMENTAL_BETWEEN, "TAG2,TAG1")))
.hasMessageContaining(
(paimon) 10/31: [spark] Add check for provider when creating table with SparkCatalog (#3326)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 508b260fd4694e6af6f102b5b6e7d95b53890bed
Author: Zouxxyy <zo...@alibaba-inc.com>
AuthorDate: Mon May 13 15:41:45 2024 +0800
[spark] Add check for provider when creating table with SparkCatalog (#3326)
---
.../main/java/org/apache/paimon/spark/SparkCatalog.java | 17 +++++++++++------
.../org/apache/paimon/spark/SparkGenericCatalog.java | 4 ----
.../apache/paimon/spark/catalog/SparkBaseCatalog.java | 5 +++++
.../scala/org/apache/paimon/spark/sql/DDLTestBase.scala | 13 +++++++++++++
4 files changed, 29 insertions(+), 10 deletions(-)
diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java
index 8f98ea91e..4e8d8eaf7 100644
--- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java
+++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java
@@ -27,7 +27,6 @@ import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaChange;
import org.apache.paimon.spark.catalog.SparkBaseCatalog;
import org.apache.paimon.table.Table;
-import org.apache.paimon.utils.Preconditions;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.catalyst.analysis.NamespaceAlreadyExistsException;
@@ -56,6 +55,7 @@ import java.util.stream.Collectors;
import static org.apache.paimon.spark.SparkCatalogOptions.DEFAULT_DATABASE;
import static org.apache.paimon.spark.SparkTypeUtils.toPaimonType;
+import static org.apache.paimon.utils.Preconditions.checkArgument;
/** Spark {@link TableCatalog} for paimon. */
public class SparkCatalog extends SparkBaseCatalog {
@@ -99,7 +99,7 @@ public class SparkCatalog extends SparkBaseCatalog {
@Override
public void createNamespace(String[] namespace, Map<String, String> metadata)
throws NamespaceAlreadyExistsException {
- Preconditions.checkArgument(
+ checkArgument(
isValidateNamespace(namespace),
"Namespace %s is not valid",
Arrays.toString(namespace));
@@ -137,7 +137,7 @@ public class SparkCatalog extends SparkBaseCatalog {
@Override
public Map<String, String> loadNamespaceMetadata(String[] namespace)
throws NoSuchNamespaceException {
- Preconditions.checkArgument(
+ checkArgument(
isValidateNamespace(namespace),
"Namespace %s is not valid",
Arrays.toString(namespace));
@@ -178,7 +178,7 @@ public class SparkCatalog extends SparkBaseCatalog {
*/
public boolean dropNamespace(String[] namespace, boolean cascade)
throws NoSuchNamespaceException {
- Preconditions.checkArgument(
+ checkArgument(
isValidateNamespace(namespace),
"Namespace %s is not valid",
Arrays.toString(namespace));
@@ -195,7 +195,7 @@ public class SparkCatalog extends SparkBaseCatalog {
@Override
public Identifier[] listTables(String[] namespace) throws NoSuchNamespaceException {
- Preconditions.checkArgument(
+ checkArgument(
isValidateNamespace(namespace),
"Missing database in namespace: %s",
Arrays.toString(namespace));
@@ -284,6 +284,11 @@ public class SparkCatalog extends SparkBaseCatalog {
Map<String, String> properties)
throws TableAlreadyExistsException, NoSuchNamespaceException {
try {
+ String provider = properties.get(TableCatalog.PROP_PROVIDER);
+ checkArgument(
+ usePaimon(provider),
+ "SparkCatalog can only create paimon table, but current provider is %s",
+ provider);
catalog.createTable(
toIdentifier(ident), toInitialSchema(schema, partitions, properties), false);
return loadTable(ident);
@@ -377,7 +382,7 @@ public class SparkCatalog extends SparkBaseCatalog {
private Schema toInitialSchema(
StructType schema, Transform[] partitions, Map<String, String> properties) {
- Preconditions.checkArgument(
+ checkArgument(
Arrays.stream(partitions)
.allMatch(
partition -> {
diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkGenericCatalog.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkGenericCatalog.java
index f8a6b203c..b947523fe 100644
--- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkGenericCatalog.java
+++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkGenericCatalog.java
@@ -324,10 +324,6 @@ public class SparkGenericCatalog extends SparkBaseCatalog implements CatalogExte
}
}
- private boolean usePaimon(String provider) {
- return provider == null || SparkSource.NAME().equalsIgnoreCase(provider);
- }
-
private TableCatalog asTableCatalog() {
return (TableCatalog) sessionCatalog;
}
diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/SparkBaseCatalog.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/SparkBaseCatalog.java
index 2f5267029..55670a594 100644
--- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/SparkBaseCatalog.java
+++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/catalog/SparkBaseCatalog.java
@@ -20,6 +20,7 @@ package org.apache.paimon.spark.catalog;
import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.spark.SparkProcedures;
+import org.apache.paimon.spark.SparkSource;
import org.apache.paimon.spark.analysis.NoSuchProcedureException;
import org.apache.paimon.spark.procedure.Procedure;
import org.apache.paimon.spark.procedure.ProcedureBuilder;
@@ -49,4 +50,8 @@ public abstract class SparkBaseCatalog
}
throw new NoSuchProcedureException(identifier);
}
+
+ public boolean usePaimon(String provider) {
+ return provider == null || SparkSource.NAME().equalsIgnoreCase(provider);
+ }
}
diff --git a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/DDLTestBase.scala b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/DDLTestBase.scala
index 9ddb83118..6fda6b26c 100644
--- a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/DDLTestBase.scala
+++ b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/DDLTestBase.scala
@@ -71,4 +71,17 @@ abstract class DDLTestBase extends PaimonSparkTestBase {
}
}
}
+
+ test("Paimon DDL: create other table with paimon SparkCatalog") {
+ withTable("paimon_tbl1", "paimon_tbl2", "parquet_tbl") {
+ spark.sql(s"CREATE TABLE paimon_tbl1 (id int) USING paimon")
+ spark.sql(s"CREATE TABLE paimon_tbl2 (id int)")
+ val error = intercept[Exception] {
+ spark.sql(s"CREATE TABLE parquet_tbl (id int) USING parquet")
+ }.getMessage
+ assert(
+ error.contains(
+ "SparkCatalog can only create paimon table, but current provider is parquet"))
+ }
+ }
}
(paimon) 22/31: [core] Fix that ignore-delete hasn't handle rowkind.field (#3365)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit e3e4303ab5e8a9686ed348b971e09b9160baa09f
Author: yuzelin <33...@users.noreply.github.com>
AuthorDate: Wed May 22 16:47:18 2024 +0800
[core] Fix that ignore-delete hasn't handle rowkind.field (#3365)
---
.../apache/paimon/table/AbstractFileStoreTable.java | 5 +++++
.../paimon/table/AppendOnlyFileStoreTable.java | 7 ++++---
.../paimon/table/PrimaryKeyFileStoreTable.java | 21 +++++++--------------
.../apache/paimon/table/sink/RowKindGenerator.java | 4 ++++
.../apache/paimon/table/sink/TableWriteImpl.java | 16 +++++++++++-----
.../paimon/flink/sink/LocalMergeOperator.java | 3 +--
.../apache/paimon/flink/BatchFileStoreITCase.java | 16 ++++++++++++++++
7 files changed, 48 insertions(+), 24 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java b/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java
index 7d4a4a7c2..9a590a0c4 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java
@@ -40,6 +40,7 @@ import org.apache.paimon.table.sink.CommitCallback;
import org.apache.paimon.table.sink.DynamicBucketRowKeyExtractor;
import org.apache.paimon.table.sink.FixedBucketRowKeyExtractor;
import org.apache.paimon.table.sink.RowKeyExtractor;
+import org.apache.paimon.table.sink.RowKindGenerator;
import org.apache.paimon.table.sink.TableCommitImpl;
import org.apache.paimon.table.sink.UnawareBucketRowKeyExtractor;
import org.apache.paimon.table.source.InnerStreamTableScan;
@@ -588,6 +589,10 @@ abstract class AbstractFileStoreTable implements FileStoreTable {
store().newTagDeletion());
}
+ protected RowKindGenerator rowKindGenerator() {
+ return RowKindGenerator.create(schema(), store().options());
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) {
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/AppendOnlyFileStoreTable.java b/paimon-core/src/main/java/org/apache/paimon/table/AppendOnlyFileStoreTable.java
index 810669f2e..f45865d01 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/AppendOnlyFileStoreTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/AppendOnlyFileStoreTable.java
@@ -147,13 +147,14 @@ class AppendOnlyFileStoreTable extends AbstractFileStoreTable {
return new TableWriteImpl<>(
writer,
createRowKeyExtractor(),
- record -> {
+ (record, rowKind) -> {
Preconditions.checkState(
- record.row().getRowKind() == RowKind.INSERT,
+ rowKind.isAdd(),
"Append only writer can not accept row with RowKind %s",
- record.row().getRowKind());
+ rowKind);
return record.row();
},
+ rowKindGenerator(),
CoreOptions.fromMap(tableSchema.options()).ignoreDelete());
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyFileStoreTable.java b/paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyFileStoreTable.java
index bf26ec31c..8c6925d73 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyFileStoreTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyFileStoreTable.java
@@ -21,7 +21,6 @@ package org.apache.paimon.table;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.KeyValue;
import org.apache.paimon.KeyValueFileStore;
-import org.apache.paimon.data.InternalRow;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.Path;
import org.apache.paimon.manifest.ManifestCacheFilter;
@@ -35,13 +34,11 @@ import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.schema.KeyValueFieldsExtractor;
import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.query.LocalTableQuery;
-import org.apache.paimon.table.sink.RowKindGenerator;
import org.apache.paimon.table.sink.TableWriteImpl;
import org.apache.paimon.table.source.InnerTableRead;
import org.apache.paimon.table.source.KeyValueTableRead;
import org.apache.paimon.table.source.MergeTreeSplitGenerator;
import org.apache.paimon.table.source.SplitGenerator;
-import org.apache.paimon.types.RowKind;
import org.apache.paimon.types.RowType;
import java.util.List;
@@ -167,21 +164,17 @@ class PrimaryKeyFileStoreTable extends AbstractFileStoreTable {
@Override
public TableWriteImpl<KeyValue> newWrite(
String commitUser, ManifestCacheFilter manifestFilter) {
- TableSchema schema = schema();
- CoreOptions options = store().options();
- RowKindGenerator rowKindGenerator = RowKindGenerator.create(schema, options);
KeyValue kv = new KeyValue();
return new TableWriteImpl<>(
store().newWrite(commitUser, manifestFilter),
createRowKeyExtractor(),
- record -> {
- InternalRow row = record.row();
- RowKind rowKind =
- rowKindGenerator == null
- ? row.getRowKind()
- : rowKindGenerator.generate(row);
- return kv.replace(record.primaryKey(), KeyValue.UNKNOWN_SEQUENCE, rowKind, row);
- },
+ (record, rowKind) ->
+ kv.replace(
+ record.primaryKey(),
+ KeyValue.UNKNOWN_SEQUENCE,
+ rowKind,
+ record.row()),
+ rowKindGenerator(),
CoreOptions.fromMap(tableSchema.options()).ignoreDelete());
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/sink/RowKindGenerator.java b/paimon-core/src/main/java/org/apache/paimon/table/sink/RowKindGenerator.java
index de55afceb..1834fdffa 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/sink/RowKindGenerator.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/sink/RowKindGenerator.java
@@ -62,4 +62,8 @@ public class RowKindGenerator {
.map(field -> new RowKindGenerator(field, schema.logicalRowType()))
.orElse(null);
}
+
+ public static RowKind getRowKind(@Nullable RowKindGenerator rowKindGenerator, InternalRow row) {
+ return rowKindGenerator == null ? row.getRowKind() : rowKindGenerator.generate(row);
+ }
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/sink/TableWriteImpl.java b/paimon-core/src/main/java/org/apache/paimon/table/sink/TableWriteImpl.java
index c03ef7e13..d7eaaa4b0 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/sink/TableWriteImpl.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/sink/TableWriteImpl.java
@@ -30,6 +30,7 @@ import org.apache.paimon.metrics.MetricRegistry;
import org.apache.paimon.operation.FileStoreWrite;
import org.apache.paimon.operation.FileStoreWrite.State;
import org.apache.paimon.table.BucketMode;
+import org.apache.paimon.types.RowKind;
import org.apache.paimon.utils.Restorable;
import javax.annotation.Nullable;
@@ -49,6 +50,7 @@ public class TableWriteImpl<T> implements InnerTableWrite, Restorable<List<State
private final FileStoreWrite<T> write;
private final KeyAndBucketExtractor<InternalRow> keyAndBucketExtractor;
private final RecordExtractor<T> recordExtractor;
+ @Nullable private final RowKindGenerator rowKindGenerator;
private final boolean ignoreDelete;
private boolean batchCommitted = false;
@@ -58,10 +60,12 @@ public class TableWriteImpl<T> implements InnerTableWrite, Restorable<List<State
FileStoreWrite<T> write,
KeyAndBucketExtractor<InternalRow> keyAndBucketExtractor,
RecordExtractor<T> recordExtractor,
+ @Nullable RowKindGenerator rowKindGenerator,
boolean ignoreDelete) {
this.write = write;
this.keyAndBucketExtractor = keyAndBucketExtractor;
this.recordExtractor = recordExtractor;
+ this.rowKindGenerator = rowKindGenerator;
this.ignoreDelete = ignoreDelete;
}
@@ -128,21 +132,23 @@ public class TableWriteImpl<T> implements InnerTableWrite, Restorable<List<State
@Nullable
public SinkRecord writeAndReturn(InternalRow row) throws Exception {
- if (ignoreDelete && row.getRowKind().isRetract()) {
+ RowKind rowKind = RowKindGenerator.getRowKind(rowKindGenerator, row);
+ if (ignoreDelete && rowKind.isRetract()) {
return null;
}
SinkRecord record = toSinkRecord(row);
- write.write(record.partition(), record.bucket(), recordExtractor.extract(record));
+ write.write(record.partition(), record.bucket(), recordExtractor.extract(record, rowKind));
return record;
}
@Nullable
public SinkRecord writeAndReturn(InternalRow row, int bucket) throws Exception {
- if (ignoreDelete && row.getRowKind().isRetract()) {
+ RowKind rowKind = RowKindGenerator.getRowKind(rowKindGenerator, row);
+ if (ignoreDelete && rowKind.isRetract()) {
return null;
}
SinkRecord record = toSinkRecord(row, bucket);
- write.write(record.partition(), bucket, recordExtractor.extract(record));
+ write.write(record.partition(), bucket, recordExtractor.extract(record, rowKind));
return record;
}
@@ -231,6 +237,6 @@ public class TableWriteImpl<T> implements InnerTableWrite, Restorable<List<State
/** Extractor to extract {@link T} from the {@link SinkRecord}. */
public interface RecordExtractor<T> {
- T extract(SinkRecord record);
+ T extract(SinkRecord record, RowKind rowKind);
}
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/LocalMergeOperator.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/LocalMergeOperator.java
index a09372443..273f38308 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/LocalMergeOperator.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/LocalMergeOperator.java
@@ -137,8 +137,7 @@ public class LocalMergeOperator extends AbstractStreamOperator<InternalRow>
recordCount++;
InternalRow row = record.getValue();
- RowKind rowKind =
- rowKindGenerator == null ? row.getRowKind() : rowKindGenerator.generate(row);
+ RowKind rowKind = RowKindGenerator.getRowKind(rowKindGenerator, row);
if (ignoreDelete && rowKind.isRetract()) {
return;
}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java
index 2758b7352..55bd886ad 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java
@@ -422,6 +422,22 @@ public class BatchFileStoreITCase extends CatalogITCaseBase {
assertThat(sql("SELECT * FROM ignore_delete")).containsExactly(Row.of(1, "B"));
}
+ @Test
+ public void testIgnoreDeleteWithRowKindField() {
+ sql(
+ "CREATE TABLE ignore_delete (pk INT PRIMARY KEY NOT ENFORCED, v STRING, kind STRING) "
+ + "WITH ('merge-engine' = 'deduplicate', 'ignore-delete' = 'true', 'bucket' = '1', 'rowkind.field' = 'kind')");
+
+ sql("INSERT INTO ignore_delete VALUES (1, 'A', '+I')");
+ assertThat(sql("SELECT * FROM ignore_delete")).containsExactly(Row.of(1, "A", "+I"));
+
+ sql("INSERT INTO ignore_delete VALUES (1, 'A', '-D')");
+ assertThat(sql("SELECT * FROM ignore_delete")).containsExactly(Row.of(1, "A", "+I"));
+
+ sql("INSERT INTO ignore_delete VALUES (1, 'B', '+I')");
+ assertThat(sql("SELECT * FROM ignore_delete")).containsExactly(Row.of(1, "B", "+I"));
+ }
+
@Test
public void testDeleteWithPkLookup() throws Exception {
sql(
(paimon) 24/31: [spark] Fix HiveMigrator hive api version compatible issue (#3381)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit f91a22029148bafe6322f533e288a59bab6d9d18
Author: xuzifu666 <12...@qq.com>
AuthorDate: Wed May 22 19:12:15 2024 +0800
[spark] Fix HiveMigrator hive api version compatible issue (#3381)
---
.../main/java/org/apache/paimon/hive/migrate/HiveMigrator.java | 9 +++++++--
1 file changed, 7 insertions(+), 2 deletions(-)
diff --git a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/migrate/HiveMigrator.java b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/migrate/HiveMigrator.java
index fd31ec3ec..513f7f665 100644
--- a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/migrate/HiveMigrator.java
+++ b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/migrate/HiveMigrator.java
@@ -216,8 +216,13 @@ public class HiveMigrator implements Migrator {
private void checkPrimaryKey() throws Exception {
PrimaryKeysRequest primaryKeysRequest = new PrimaryKeysRequest(sourceDatabase, sourceTable);
- if (!client.getPrimaryKeys(primaryKeysRequest).isEmpty()) {
- throw new IllegalArgumentException("Can't migrate primary key table yet.");
+ try {
+ if (!client.getPrimaryKeys(primaryKeysRequest).isEmpty()) {
+ throw new IllegalArgumentException("Can't migrate primary key table yet.");
+ }
+ } catch (Exception e) {
+ LOG.warn(
+ "Your Hive version is low which not support get_primary_keys, skip primary key check firstly!");
}
}
(paimon) 04/31: [cdc] Fixed when the order of the same field differs, it is considered a schema change. (#3314)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 5802088285f12f70941602f846a8b83e5523b327
Author: chenxi0599 <ch...@gmail.com>
AuthorDate: Fri May 10 13:56:38 2024 +0800
[cdc] Fixed when the order of the same field differs, it is considered a schema change. (#3314)
---
.../apache/paimon/flink/sink/cdc/RichEventParser.java | 17 ++++++++++++++++-
1 file changed, 16 insertions(+), 1 deletion(-)
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/RichEventParser.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/RichEventParser.java
index 756875781..10dbdcc8d 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/RichEventParser.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/RichEventParser.java
@@ -45,7 +45,10 @@ public class RichEventParser implements EventParser<RichCdcRecord> {
.forEach(
dataField -> {
DataField previous = previousDataFields.get(dataField.name());
- if (!Objects.equals(previous, dataField)) {
+ // When the order of the same field is different, its ID may also be
+ // different,
+ // so the comparison should not include the ID.
+ if (!dataFieldEqualsIgnoreId(previous, dataField)) {
previousDataFields.put(dataField.name(), dataField);
change.add(dataField);
}
@@ -53,6 +56,18 @@ public class RichEventParser implements EventParser<RichCdcRecord> {
return change;
}
+ private boolean dataFieldEqualsIgnoreId(DataField dataField1, DataField dataField2) {
+ if (dataField1 == dataField2) {
+ return true;
+ } else if (dataField1 != null && dataField2 != null) {
+ return Objects.equals(dataField1.name(), dataField2.name())
+ && Objects.equals(dataField1.type(), dataField2.type())
+ && Objects.equals(dataField1.description(), dataField2.description());
+ } else {
+ return false;
+ }
+ }
+
@Override
public List<CdcRecord> parseRecords() {
if (record.hasPayload()) {
(paimon) 03/31: [doc] add some descriptions about data writing in basic-concepts (#3301)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 64e031343f7fdd80d2ac38b4d761ab7c149277c6
Author: Aiden <12...@users.noreply.github.com>
AuthorDate: Tue May 7 14:21:25 2024 +0800
[doc] add some descriptions about data writing in basic-concepts (#3301)
---
docs/content/concepts/basic-concepts.md | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/docs/content/concepts/basic-concepts.md b/docs/content/concepts/basic-concepts.md
index 0c57e2e34..3213ca429 100644
--- a/docs/content/concepts/basic-concepts.md
+++ b/docs/content/concepts/basic-concepts.md
@@ -67,7 +67,7 @@ By partitioning, users can efficiently operate on a slice of records in the tabl
## Consistency Guarantees
Paimon writers use two-phase commit protocol to atomically commit a batch of records to the table. Each commit produces
-at most two [snapshots]({{< ref "concepts/basic-concepts#snapshot" >}}) at commit time.
+at most two [snapshots]({{< ref "concepts/basic-concepts#snapshot" >}}) at commit time. It depends on the incremental write and compaction strategy. If only incremental writes are performed without triggering a compaction operation, only an incremental snapshot will be created. If a compaction operation is triggered, an incremental snapshot and a compacted snapshot will be created.
For any two writers modifying a table at the same time, as long as they do not modify the same partition, their commits
can occur in parallel. If they modify the same partition, only snapshot isolation is guaranteed. That is, the final table
(paimon) 31/31: [hive] Fix flink not synchronizing to hive metastore when deleting table partitions using hive catalog (#3411)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit cfcc72d7d86e36a0a505023d500a663f23f6dfe0
Author: Kerwin <37...@users.noreply.github.com>
AuthorDate: Wed May 29 10:34:34 2024 +0800
[hive] Fix flink not synchronizing to hive metastore when deleting table partitions using hive catalog (#3411)
---
.../apache/paimon/table/PrimaryKeyTableUtils.java | 8 +-------
.../java/org/apache/paimon/hive/HiveCatalog.java | 20 ++++++++++++++++++++
.../apache/paimon/hive/HiveCatalogITCaseBase.java | 13 +++++++++++++
3 files changed, 34 insertions(+), 7 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyTableUtils.java b/paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyTableUtils.java
index 572e488c6..b3dbbdd29 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyTableUtils.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/PrimaryKeyTableUtils.java
@@ -45,13 +45,7 @@ public class PrimaryKeyTableUtils {
public static List<DataField> addKeyNamePrefix(List<DataField> keyFields) {
return keyFields.stream()
- .map(
- f ->
- new DataField(
- f.id(),
- KEY_FIELD_PREFIX + f.name(),
- f.type(),
- f.description()))
+ .map(f -> f.newName(KEY_FIELD_PREFIX + f.name()))
.collect(Collectors.toList());
}
diff --git a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java
index 0d71e70a6..7da826e3f 100644
--- a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java
+++ b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java
@@ -72,6 +72,7 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
@@ -267,6 +268,25 @@ public class HiveCatalog extends AbstractCatalog {
}
}
+ @Override
+ public void dropPartition(Identifier identifier, Map<String, String> partitionSpec)
+ throws TableNotExistException {
+ TableSchema tableSchema = getDataTableSchema(identifier);
+ if (!tableSchema.partitionKeys().isEmpty()
+ && new CoreOptions(tableSchema.options()).partitionedTableInMetastore()) {
+ try {
+ // Do not close client, it is for HiveCatalog
+ @SuppressWarnings("resource")
+ HiveMetastoreClient metastoreClient =
+ new HiveMetastoreClient(identifier, tableSchema, client);
+ metastoreClient.deletePartition(new LinkedHashMap<>(partitionSpec));
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ super.dropPartition(identifier, partitionSpec);
+ }
+
private Map<String, String> convertToProperties(Database database) {
Map<String, String> properties = new HashMap<>(database.getParameters());
if (database.getLocationUri() != null) {
diff --git a/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/HiveCatalogITCaseBase.java b/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/HiveCatalogITCaseBase.java
index e478d44cd..4e6fba9e7 100644
--- a/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/HiveCatalogITCaseBase.java
+++ b/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/HiveCatalogITCaseBase.java
@@ -950,6 +950,19 @@ public abstract class HiveCatalogITCaseBase {
.containsExactlyInAnyOrder("1\t10", "2\t20");
}
+ @Test
+ public void testDropPartitionsToMetastore() throws Exception {
+ prepareTestAddPartitionsToMetastore();
+
+ // drop partition
+ tEnv.executeSql(
+ "ALTER TABLE t DROP PARTITION (ptb = '1a', pta = 1), PARTITION (ptb = '1b', pta = 1)")
+ .await();
+ assertThat(hiveShell.executeQuery("show partitions t"))
+ .containsExactlyInAnyOrder(
+ "ptb=2a/pta=2", "ptb=2b/pta=2", "ptb=3a/pta=3", "ptb=3b/pta=3");
+ }
+
@Test
public void testAddPartitionsForTag() throws Exception {
tEnv.executeSql(
(paimon) 13/31: [core] Add JsonIgnoreProperties to all JSON class
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 51103e0437aa66d8f7d13625d0e1b8a3d8090845
Author: Jingsong <ji...@gmail.com>
AuthorDate: Tue May 14 09:58:57 2024 +0800
[core] Add JsonIgnoreProperties to all JSON class
---
.../src/main/java/org/apache/paimon/Changelog.java | 2 +
.../java/org/apache/paimon/consumer/Consumer.java | 2 +
.../java/org/apache/paimon/stats/ColStats.java | 2 +
.../java/org/apache/paimon/stats/Statistics.java | 2 +
.../src/main/java/org/apache/paimon/tag/Tag.java | 2 +
.../cdc/format/debezium/CdcDebeziumSchema.java | 91 ----------------------
.../action/cdc/mysql/format/DebeziumEvent.java | 2 +
7 files changed, 12 insertions(+), 91 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/Changelog.java b/paimon-core/src/main/java/org/apache/paimon/Changelog.java
index 2ab10a037..8c6295b44 100644
--- a/paimon-core/src/main/java/org/apache/paimon/Changelog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/Changelog.java
@@ -23,6 +23,7 @@ import org.apache.paimon.fs.Path;
import org.apache.paimon.utils.JsonSerdeUtil;
import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
import javax.annotation.Nullable;
@@ -36,6 +37,7 @@ import java.util.Map;
* one source:
* <li>The changelog file. Eg: from the changelog-producer = 'input'
*/
+@JsonIgnoreProperties(ignoreUnknown = true)
public class Changelog extends Snapshot {
public Changelog(Snapshot snapshot) {
diff --git a/paimon-core/src/main/java/org/apache/paimon/consumer/Consumer.java b/paimon-core/src/main/java/org/apache/paimon/consumer/Consumer.java
index f1574f199..ffa63ece3 100644
--- a/paimon-core/src/main/java/org/apache/paimon/consumer/Consumer.java
+++ b/paimon-core/src/main/java/org/apache/paimon/consumer/Consumer.java
@@ -24,6 +24,7 @@ import org.apache.paimon.utils.JsonSerdeUtil;
import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonGetter;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
import java.io.IOException;
@@ -31,6 +32,7 @@ import java.io.UncheckedIOException;
import java.util.Optional;
/** Consumer which contains next snapshot. */
+@JsonIgnoreProperties(ignoreUnknown = true)
public class Consumer {
private static final String FIELD_NEXT_SNAPSHOT = "nextSnapshot";
diff --git a/paimon-core/src/main/java/org/apache/paimon/stats/ColStats.java b/paimon-core/src/main/java/org/apache/paimon/stats/ColStats.java
index 35baf1a5b..1b376c356 100644
--- a/paimon-core/src/main/java/org/apache/paimon/stats/ColStats.java
+++ b/paimon-core/src/main/java/org/apache/paimon/stats/ColStats.java
@@ -26,6 +26,7 @@ import org.apache.paimon.utils.JsonSerdeUtil;
import org.apache.paimon.utils.OptionalUtils;
import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
@@ -50,6 +51,7 @@ import java.util.OptionalLong;
* @param <T> col internal data type
*/
@Experimental
+@JsonIgnoreProperties(ignoreUnknown = true)
public class ColStats<T> {
private static final String FIELD_COL_ID = "colId";
diff --git a/paimon-core/src/main/java/org/apache/paimon/stats/Statistics.java b/paimon-core/src/main/java/org/apache/paimon/stats/Statistics.java
index 60a121cd9..5d9403044 100644
--- a/paimon-core/src/main/java/org/apache/paimon/stats/Statistics.java
+++ b/paimon-core/src/main/java/org/apache/paimon/stats/Statistics.java
@@ -27,6 +27,7 @@ import org.apache.paimon.utils.JsonSerdeUtil;
import org.apache.paimon.utils.OptionalUtils;
import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
@@ -48,6 +49,7 @@ import java.util.OptionalLong;
* </ul>
*/
@Experimental
+@JsonIgnoreProperties(ignoreUnknown = true)
public class Statistics {
// ID of the snapshot this statistics collected from
diff --git a/paimon-core/src/main/java/org/apache/paimon/tag/Tag.java b/paimon-core/src/main/java/org/apache/paimon/tag/Tag.java
index 938bb417d..f1ac879d3 100644
--- a/paimon-core/src/main/java/org/apache/paimon/tag/Tag.java
+++ b/paimon-core/src/main/java/org/apache/paimon/tag/Tag.java
@@ -25,6 +25,7 @@ import org.apache.paimon.utils.JsonSerdeUtil;
import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonGetter;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
@@ -38,6 +39,7 @@ import java.util.Map;
import java.util.Objects;
/** Snapshot with tagCreateTime and tagTimeRetained. */
+@JsonIgnoreProperties(ignoreUnknown = true)
public class Tag extends Snapshot {
private static final String FIELD_TAG_CREATE_TIME = "tagCreateTime";
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/debezium/CdcDebeziumSchema.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/debezium/CdcDebeziumSchema.java
deleted file mode 100644
index 64f3a1267..000000000
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/debezium/CdcDebeziumSchema.java
+++ /dev/null
@@ -1,91 +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.action.cdc.format.debezium;
-
-import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
-import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonGetter;
-import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
-import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
-
-import java.util.List;
-
-/** Cdc Debezium Schema Entity. */
-@JsonIgnoreProperties(ignoreUnknown = true)
-public class CdcDebeziumSchema {
-
- private static final String FIELD_FIELDS = "fields";
- private static final String FIELD_NAME = "name";
- private static final String FIELD_TYPE = "type";
- private static final String FIELD_FIELD = "field";
- private static final String FIELD_OPTIONAL = "optional";
-
- @JsonProperty(FIELD_FIELD)
- private final String field;
-
- @JsonProperty(FIELD_TYPE)
- private final String type;
-
- @JsonProperty(FIELD_OPTIONAL)
- private final Boolean optional;
-
- @JsonProperty(FIELD_FIELDS)
- private final List<CdcDebeziumSchema> fields;
-
- @JsonProperty(FIELD_NAME)
- private final String name;
-
- @JsonCreator
- public CdcDebeziumSchema(
- @JsonProperty(FIELD_FIELD) String field,
- @JsonProperty(FIELD_TYPE) String type,
- @JsonProperty(FIELD_OPTIONAL) Boolean optional,
- @JsonProperty(FIELD_FIELDS) List<CdcDebeziumSchema> fields,
- @JsonProperty(FIELD_NAME) String name) {
- this.field = field;
- this.type = type;
- this.optional = optional;
- this.fields = fields;
- this.name = name;
- }
-
- @JsonGetter(FIELD_FIELD)
- public String field() {
- return field;
- }
-
- @JsonGetter(FIELD_TYPE)
- public String type() {
- return type;
- }
-
- @JsonGetter(FIELD_OPTIONAL)
- public Boolean optional() {
- return optional;
- }
-
- @JsonGetter(FIELD_FIELDS)
- public List<CdcDebeziumSchema> fields() {
- return fields;
- }
-
- @JsonGetter(FIELD_NAME)
- public String name() {
- return name;
- }
-}
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/format/DebeziumEvent.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/format/DebeziumEvent.java
index ebc1bf1b5..afaa4cb0a 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/format/DebeziumEvent.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/mysql/format/DebeziumEvent.java
@@ -21,6 +21,7 @@ package org.apache.paimon.flink.action.cdc.mysql.format;
import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonGetter;
import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnore;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.JsonNode;
@@ -36,6 +37,7 @@ import java.util.function.Predicate;
import java.util.stream.Collectors;
/** Debezium Event Records Entity. */
+@JsonIgnoreProperties(ignoreUnknown = true)
public class DebeziumEvent {
private static final String FIELD_PAYLOAD = "payload";
(paimon) 25/31: [flink] Cleanup QueryService file after job is down (#3382)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 6355d13a0584ea8bd8d90bb0649e5ab0a84e5f19
Author: Aitozi <ai...@apache.org>
AuthorDate: Wed May 22 19:13:04 2024 +0800
[flink] Cleanup QueryService file after job is down (#3382)
---
.../paimon/flink/service/QueryAddressRegister.java | 14 ++++++----
.../paimon/flink/RemoteLookupJoinITCase.java | 31 ++++++++++++++++++++++
2 files changed, 40 insertions(+), 5 deletions(-)
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/service/QueryAddressRegister.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/service/QueryAddressRegister.java
index ed883d9eb..df3cf7abf 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/service/QueryAddressRegister.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/service/QueryAddressRegister.java
@@ -35,13 +35,13 @@ import static org.apache.paimon.service.ServiceManager.PRIMARY_KEY_LOOKUP;
/** Operator for address server to register addresses to {@link ServiceManager}. */
public class QueryAddressRegister extends RichSinkFunction<InternalRow> {
- private final Table table;
+ private final ServiceManager serviceManager;
private transient int numberExecutors;
private transient TreeMap<Integer, InetSocketAddress> executors;
public QueryAddressRegister(Table table) {
- this.table = table;
+ this.serviceManager = ((FileStoreTable) table).store().newServiceManager();
}
@Override
@@ -67,10 +67,14 @@ public class QueryAddressRegister extends RichSinkFunction<InternalRow> {
executors.put(executorId, new InetSocketAddress(hostname, port));
if (executors.size() == numberExecutors) {
- FileStoreTable storeTable = (FileStoreTable) table;
- ServiceManager manager = storeTable.store().newServiceManager();
- manager.resetService(
+ serviceManager.resetService(
PRIMARY_KEY_LOOKUP, executors.values().toArray(new InetSocketAddress[0]));
}
}
+
+ @Override
+ public void close() throws Exception {
+ super.close();
+ serviceManager.deleteService(PRIMARY_KEY_LOOKUP);
+ }
}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/RemoteLookupJoinITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/RemoteLookupJoinITCase.java
index 9e7aac281..2d8d072d1 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/RemoteLookupJoinITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/RemoteLookupJoinITCase.java
@@ -22,6 +22,7 @@ import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.disk.IOManager;
import org.apache.paimon.flink.query.RemoteTableQuery;
+import org.apache.paimon.flink.service.QueryService;
import org.apache.paimon.service.ServiceManager;
import org.apache.paimon.service.network.stats.DisabledServiceRequestStats;
import org.apache.paimon.service.server.KvQueryServer;
@@ -34,6 +35,8 @@ import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.table.sink.CommitMessageImpl;
import org.apache.paimon.utils.BlockingIterator;
+import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.util.CloseableIterator;
import org.junit.jupiter.api.Test;
@@ -131,6 +134,34 @@ public class RemoteLookupJoinITCase extends CatalogITCaseBase {
proxy.close();
}
+ @Test
+ public void testServiceFileCleaned() throws Exception {
+ sql(
+ "CREATE TABLE DIM (k INT PRIMARY KEY NOT ENFORCED, v INT) WITH ('bucket' = '2', 'continuous.discovery-interval' = '1ms')");
+ JobClient client = queryService(paimonTable("DIM"));
+
+ RemoteTableQuery query = new RemoteTableQuery(paimonTable("DIM"));
+
+ sql("INSERT INTO DIM VALUES (1, 11), (2, 22), (3, 33), (4, 44)");
+ Thread.sleep(2000);
+
+ assertThat(query.lookup(row(), 0, row(1)))
+ .isNotNull()
+ .extracting(r -> r.getInt(1))
+ .isEqualTo(11);
+
+ client.cancel().get();
+ query.close();
+ ServiceManager serviceManager = paimonTable("DIM").store().newServiceManager();
+ assertThat(serviceManager.service(PRIMARY_KEY_LOOKUP).isPresent()).isFalse();
+ }
+
+ private JobClient queryService(FileStoreTable table) throws Exception {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ QueryService.build(env, table, 2);
+ return env.executeAsync();
+ }
+
private ServiceProxy launchQueryServer(String tableName) throws Throwable {
FileStoreTable table = (FileStoreTable) paimonTable(tableName);
LocalTableQuery query = table.newLocalTableQuery().withIOManager(IOManager.create(path));
(paimon) 15/31: [core] Avoid potential NPE in FileStoreCommitImpl and TagAutoCreation
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 688d12adfc9c4d91d9be5edc761ad59e47ceb664
Author: Jingsong <ji...@gmail.com>
AuthorDate: Tue May 14 17:02:51 2024 +0800
[core] Avoid potential NPE in FileStoreCommitImpl and TagAutoCreation
---
paimon-core/src/main/java/org/apache/paimon/Snapshot.java | 1 +
.../java/org/apache/paimon/operation/FileStoreCommitImpl.java | 5 ++++-
.../src/main/java/org/apache/paimon/tag/TagAutoCreation.java | 8 ++++++--
3 files changed, 11 insertions(+), 3 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/Snapshot.java b/paimon-core/src/main/java/org/apache/paimon/Snapshot.java
index 20d414092..6c2656d39 100644
--- a/paimon-core/src/main/java/org/apache/paimon/Snapshot.java
+++ b/paimon-core/src/main/java/org/apache/paimon/Snapshot.java
@@ -316,6 +316,7 @@ public class Snapshot {
}
@JsonGetter(FIELD_LOG_OFFSETS)
+ @Nullable
public Map<Integer, Long> logOffsets() {
return logOffsets;
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
index be5255a36..dab336b81 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
@@ -793,7 +793,10 @@ public class FileStoreCommitImpl implements FileStoreCommit {
oldMetas.addAll(previousManifests);
// read the last snapshot to complete the bucket's offsets when logOffsets does not
// contain all buckets
- latestSnapshot.logOffsets().forEach(logOffsets::putIfAbsent);
+ Map<Integer, Long> latestLogOffsets = latestSnapshot.logOffsets();
+ if (latestLogOffsets != null) {
+ latestLogOffsets.forEach(logOffsets::putIfAbsent);
+ }
Long latestWatermark = latestSnapshot.watermark();
if (latestWatermark != null) {
currentWatermark =
diff --git a/paimon-core/src/main/java/org/apache/paimon/tag/TagAutoCreation.java b/paimon-core/src/main/java/org/apache/paimon/tag/TagAutoCreation.java
index 60f113761..af7947925 100644
--- a/paimon-core/src/main/java/org/apache/paimon/tag/TagAutoCreation.java
+++ b/paimon-core/src/main/java/org/apache/paimon/tag/TagAutoCreation.java
@@ -109,10 +109,14 @@ public class TagAutoCreation {
return false;
}
+ Long watermark = latestSnapshot.watermark();
+ if (watermark == null) {
+ return false;
+ }
+
LocalDateTime snapshotTime =
LocalDateTime.ofInstant(
- Instant.ofEpochMilli(latestSnapshot.watermark()),
- ZoneId.systemDefault());
+ Instant.ofEpochMilli(watermark), ZoneId.systemDefault());
return isAfterOrEqual(LocalDateTime.now().minus(idlenessTimeout), snapshotTime);
} else if (timeExtractor instanceof ProcessTimeExtractor) {
(paimon) 12/31: [doc] Optimize useless message in dedicated-compaction
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit b8b6e8664ecbc6ee1bb858e94f3bcdd0e7bf5e4d
Author: Jingsong <ji...@gmail.com>
AuthorDate: Mon May 13 21:10:19 2024 +0800
[doc] Optimize useless message in dedicated-compaction
---
docs/content/maintenance/dedicated-compaction.md | 8 +++++---
1 file changed, 5 insertions(+), 3 deletions(-)
diff --git a/docs/content/maintenance/dedicated-compaction.md b/docs/content/maintenance/dedicated-compaction.md
index f000849bf..ae975f4c9 100644
--- a/docs/content/maintenance/dedicated-compaction.md
+++ b/docs/content/maintenance/dedicated-compaction.md
@@ -45,10 +45,12 @@ streaming jobs to write records to a `'partial-update'` table. Please refer to t
## Dedicated Compaction Job
-By default, Paimon writers will perform compaction as needed during writing records. This is sufficient for most use cases, but there are two downsides:
+By default, Paimon writers will perform compaction as needed during writing records. This is sufficient for most use cases.
-* This may result in unstable write throughput because throughput might temporarily drop when performing a compaction.
-* Compaction will mark some data files as "deleted" (not really deleted, see [expiring snapshots]({{< ref "maintenance/manage-snapshots#expiring-snapshots" >}}) for more info). If multiple writers mark the same file, a conflict will occur when committing the changes. Paimon will automatically resolve the conflict, but this may result in job restarts.
+Compaction will mark some data files as "deleted" (not really deleted, see
+[expiring snapshots]({{< ref "maintenance/manage-snapshots#expiring-snapshots" >}}) for more info).
+If multiple writers mark the same file, a conflict will occur when committing the changes. Paimon will automatically
+resolve the conflict, but this may result in job restarts.
To avoid these downsides, users can also choose to skip compactions in writers, and run a dedicated job only for compaction. As compactions are performed only by the dedicated job, writers can continuously write records without pausing and no conflicts will ever occur.
(paimon) 29/31: [doc] Update managed memory guide for write buffer managing (#3417)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit f3e67b7b311db773c6b0fd74b82f7e1a5f1cffe4
Author: GuojunLi <li...@bytedance.com>
AuthorDate: Tue May 28 20:07:54 2024 +0800
[doc] Update managed memory guide for write buffer managing (#3417)
---
docs/content/maintenance/write-performance.md | 4 ++++
1 file changed, 4 insertions(+)
diff --git a/docs/content/maintenance/write-performance.md b/docs/content/maintenance/write-performance.md
index 0129c07a0..d69cdc1a9 100644
--- a/docs/content/maintenance/write-performance.md
+++ b/docs/content/maintenance/write-performance.md
@@ -249,6 +249,10 @@ If your Flink job does not rely on state, please avoid using managed memory, whi
```shell
taskmanager.memory.managed.size=1m
```
+Or you can use Flink managed memory for your write buffer to avoid OOM, set table property:
+```shell
+sink.use-managed-memory-allocator=true
+```
## Commit Memory
(paimon) 08/31: [spark] Fix drop database cascade when database is not empy with SparkGenericCatalog (#3324)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 3c75ca773a6cb4f337adde7705b2b822ef8ff7ad
Author: Zouxxyy <zo...@alibaba-inc.com>
AuthorDate: Mon May 13 11:56:46 2024 +0800
[spark] Fix drop database cascade when database is not empy with SparkGenericCatalog (#3324)
---
.../apache/paimon/spark/SparkGenericCatalog.java | 5 ++++
.../apache/paimon/spark/PaimonHiveTestBase.scala | 10 ++++----
.../apache/paimon/spark/PaimonSparkTestBase.scala | 8 +++++--
.../spark/sql/DDLWithHiveCatalogTestBase.scala | 23 ++++++++++++++----
.../paimon/spark/sql/SparkVersionSupport.scala | 27 ++++++++++++++++++++++
5 files changed, 63 insertions(+), 10 deletions(-)
diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkGenericCatalog.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkGenericCatalog.java
index cf8ee38ab..f8a6b203c 100644
--- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkGenericCatalog.java
+++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkGenericCatalog.java
@@ -132,6 +132,11 @@ public class SparkGenericCatalog extends SparkBaseCatalog implements CatalogExte
@Override
public boolean dropNamespace(String[] namespace, boolean cascade)
throws NoSuchNamespaceException, NonEmptyNamespaceException {
+ if (namespace.length == 1 && namespaceExists(namespace) && cascade) {
+ for (Identifier table : listTables(namespace)) {
+ dropTable(table);
+ }
+ }
return asNamespaceCatalog().dropNamespace(namespace, cascade);
}
diff --git a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/PaimonHiveTestBase.scala b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/PaimonHiveTestBase.scala
index a87502b22..842147615 100644
--- a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/PaimonHiveTestBase.scala
+++ b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/PaimonHiveTestBase.scala
@@ -34,6 +34,8 @@ class PaimonHiveTestBase extends PaimonSparkTestBase {
protected lazy val testHiveMetastore: TestHiveMetastore = new TestHiveMetastore
+ protected val sparkCatalogName: String = "spark_catalog"
+
protected val paimonHiveCatalogName: String = "paimon_hive"
protected val hiveDbName: String = "test_hive"
@@ -46,7 +48,7 @@ class PaimonHiveTestBase extends PaimonSparkTestBase {
super.sparkConf
.set("spark.sql.warehouse.dir", tempHiveDBDir.getCanonicalPath)
.set("spark.sql.catalogImplementation", "hive")
- .set("spark.sql.catalog.spark_catalog", "org.apache.paimon.spark.SparkGenericCatalog")
+ .set(s"spark.sql.catalog.$sparkCatalogName", "org.apache.paimon.spark.SparkGenericCatalog")
.set(s"spark.sql.catalog.$paimonHiveCatalogName", classOf[SparkCatalog].getName)
.set(s"spark.sql.catalog.$paimonHiveCatalogName.metastore", "hive")
.set(s"spark.sql.catalog.$paimonHiveCatalogName.warehouse", tempHiveDBDir.getCanonicalPath)
@@ -56,13 +58,13 @@ class PaimonHiveTestBase extends PaimonSparkTestBase {
override protected def beforeAll(): Unit = {
testHiveMetastore.start(hivePort)
super.beforeAll()
- spark.sql(s"USE spark_catalog")
+ spark.sql(s"USE $sparkCatalogName")
spark.sql(s"CREATE DATABASE IF NOT EXISTS $hiveDbName")
}
override protected def afterAll(): Unit = {
try {
- spark.sql(s"USE spark_catalog")
+ spark.sql(s"USE $sparkCatalogName")
spark.sql("USE default")
spark.sql(s"DROP DATABASE $hiveDbName CASCADE")
} finally {
@@ -73,7 +75,7 @@ class PaimonHiveTestBase extends PaimonSparkTestBase {
/** Default is spark_catalog */
override protected def beforeEach(): Unit = {
- spark.sql(s"USE spark_catalog")
+ spark.sql(s"USE $sparkCatalogName")
spark.sql(s"USE $hiveDbName")
}
}
diff --git a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/PaimonSparkTestBase.scala b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/PaimonSparkTestBase.scala
index 0e79d1080..6ccf08016 100644
--- a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/PaimonSparkTestBase.scala
+++ b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/PaimonSparkTestBase.scala
@@ -22,7 +22,7 @@ import org.apache.paimon.catalog.{Catalog, CatalogContext, CatalogFactory, Ident
import org.apache.paimon.options.Options
import org.apache.paimon.spark.catalog.Catalogs
import org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions
-import org.apache.paimon.spark.sql.WithTableOptions
+import org.apache.paimon.spark.sql.{SparkVersionSupport, WithTableOptions}
import org.apache.paimon.table.FileStoreTable
import org.apache.spark.SparkConf
@@ -39,7 +39,11 @@ import java.io.File
import scala.util.Random
-class PaimonSparkTestBase extends QueryTest with SharedSparkSession with WithTableOptions {
+class PaimonSparkTestBase
+ extends QueryTest
+ with SharedSparkSession
+ with WithTableOptions
+ with SparkVersionSupport {
protected lazy val tempDBDir: File = Utils.createTempDir
diff --git a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/DDLWithHiveCatalogTestBase.scala b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/DDLWithHiveCatalogTestBase.scala
index a4ef0a55d..c03f61bf6 100644
--- a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/DDLWithHiveCatalogTestBase.scala
+++ b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/DDLWithHiveCatalogTestBase.scala
@@ -26,7 +26,7 @@ import org.junit.jupiter.api.Assertions
abstract class DDLWithHiveCatalogTestBase extends PaimonHiveTestBase {
test("Paimon DDL with hive catalog: create database with location and comment") {
- Seq("spark_catalog", paimonHiveCatalogName).foreach {
+ Seq(sparkCatalogName, paimonHiveCatalogName).foreach {
catalogName =>
spark.sql(s"USE $catalogName")
withTempDir {
@@ -54,7 +54,7 @@ abstract class DDLWithHiveCatalogTestBase extends PaimonHiveTestBase {
}
test("Paimon DDL with hive catalog: create database with props") {
- Seq("spark_catalog", paimonHiveCatalogName).foreach {
+ Seq(sparkCatalogName, paimonHiveCatalogName).foreach {
catalogName =>
spark.sql(s"USE $catalogName")
withDatabase("paimon_db") {
@@ -69,7 +69,7 @@ abstract class DDLWithHiveCatalogTestBase extends PaimonHiveTestBase {
test("Paimon DDL with hive catalog: set default database") {
var reusedSpark = spark
- Seq("paimon", "spark_catalog", paimonHiveCatalogName).foreach {
+ Seq("paimon", sparkCatalogName, paimonHiveCatalogName).foreach {
catalogName =>
{
val dbName = s"${catalogName}_default_db"
@@ -89,7 +89,7 @@ abstract class DDLWithHiveCatalogTestBase extends PaimonHiveTestBase {
.config(s"spark.sql.catalog.$catalogName.defaultDatabase", dbName)
.getOrCreate()
- if (catalogName.equals("spark_catalog") && !supportDefaultDatabaseWithSessionCatalog) {
+ if (catalogName.equals(sparkCatalogName) && !supportDefaultDatabaseWithSessionCatalog) {
checkAnswer(reusedSpark.sql("show tables").select("tableName"), Nil)
reusedSpark.sql(s"use $dbName")
}
@@ -102,6 +102,21 @@ abstract class DDLWithHiveCatalogTestBase extends PaimonHiveTestBase {
reusedSpark.stop()
}
+ test("Paimon DDL with hive catalog: drop database cascade which contains paimon table") {
+ // Spark supports DROP DATABASE CASCADE since 3.3
+ if (gteqSpark3_3) {
+ Seq(sparkCatalogName, paimonHiveCatalogName).foreach {
+ catalogName =>
+ spark.sql(s"USE $catalogName")
+ spark.sql(s"CREATE DATABASE paimon_db")
+ spark.sql(s"USE paimon_db")
+ spark.sql(s"CREATE TABLE paimon_tbl (id int, name string, dt string) using paimon")
+ spark.sql(s"USE default")
+ spark.sql(s"DROP DATABASE paimon_db CASCADE")
+ }
+ }
+ }
+
def supportDefaultDatabaseWithSessionCatalog = true
def getDatabaseLocation(dbName: String): String = {
diff --git a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/SparkVersionSupport.scala b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/SparkVersionSupport.scala
new file mode 100644
index 000000000..c8f5711a6
--- /dev/null
+++ b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/SparkVersionSupport.scala
@@ -0,0 +1,27 @@
+/*
+ * 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.sql
+
+import org.apache.spark.SPARK_VERSION
+
+trait SparkVersionSupport {
+ lazy val sparkVersion: String = SPARK_VERSION
+
+ lazy val gteqSpark3_3: Boolean = sparkVersion >= "3.3"
+}
(paimon) 14/31: [doc] Add Specification Page
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit a088a044c230e869931891bf16d7e86761f854e6
Author: Jingsong <ji...@gmail.com>
AuthorDate: Mon May 13 21:08:07 2024 +0800
[doc] Add Specification Page
---
docs/content/concepts/specification.md | 253 +++++++++++++++++++++++++++++++++
1 file changed, 253 insertions(+)
diff --git a/docs/content/concepts/specification.md b/docs/content/concepts/specification.md
new file mode 100644
index 000000000..aae8970b6
--- /dev/null
+++ b/docs/content/concepts/specification.md
@@ -0,0 +1,253 @@
+---
+title: "Specification"
+weight: 4
+type: docs
+aliases:
+- /concepts/spec.html
+---
+<!--
+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.
+-->
+
+# Specification
+
+This is the specification for the Paimon table format, this document standardizes the underlying file structure and
+design of Paimon.
+
+{{< img src="/img/file-layout.png">}}
+
+## Terms
+
+- Schema: fields, primary keys definition, partition keys definition and options.
+- Snapshot: the entrance to all data committed at some specific time point.
+- Manifest list: includes several manifest files.
+- Manifest: includes several data files or changelog files.
+- Data File: contains incremental records.
+- Changelog File: contains records produced by changelog-producer.
+- Global Index: index for a bucket or partition.
+- Data File Index: index for a data file.
+
+Run Flink SQL with Paimon:
+
+```sql
+CREATE CATALOG my_catalog WITH (
+ 'type' = 'paimon',
+ 'warehouse' = '/your/path'
+);
+USE CATALOG my_catalog;
+
+CREATE TABLE my_table (
+ k INT PRIMARY KEY NOT ENFORCED,
+ f0 INT,
+ f1 STRING
+);
+
+INSERT INTO t VALUES (1, 11, '111');
+```
+
+Take a look to the disk:
+
+```shell
+warehouse
+└── default.db
+ └── my_table
+ ├── bucket-0
+ │ └── data-59f60cb9-44af-48cc-b5ad-59e85c663c8f-0.orc
+ ├── index
+ │ └── index-5625e6d9-dd44-403b-a738-2b6ea92e20f1-0
+ ├── manifest
+ │ ├── index-manifest-5d670043-da25-4265-9a26-e31affc98039-0
+ │ ├── manifest-6758823b-2010-4d06-aef0-3b1b597723d6-0
+ │ ├── manifest-list-9f856d52-5b33-4c10-8933-a0eddfaa25bf-0
+ │ └── manifest-list-9f856d52-5b33-4c10-8933-a0eddfaa25bf-1
+ ├── schema
+ │ └── schema-0
+ └── snapshot
+ ├── EARLIEST
+ ├── LATEST
+ └── snapshot-1
+```
+
+## Schema
+
+The version of the schema file starts from 0 and currently retains all versions of the schema. There may be old files
+that rely on the old schema version, so its deletion should be done with caution.
+
+Schema File is JSON, it includes:
+
+1. fields: data field list, data field contains `id`, `name`, `type`, field id is used to support schema evolution.
+2. partitionKeys: partition definition of the table, it cannot be modified.
+3. primaryKeys: primary key definition of the table, it cannot be modified.
+4. options: options of the table, including a lot of capabilities and optimizations.
+
+## Snapshot
+
+Each commit generates a snapshot file, and the version of the snapshot file starts from 1 and must be continuous.
+`EARLIEST` and `LATEST` are hint files at the beginning and end of the snapshot list, and they can be inaccurate.
+When hint files are inaccurate, the read will scan all snapshot files to determine the beginning and end.
+
+Writing commit will preempt the next snapshot id, and once the snapshot file is successfully written, this commit will
+be visible.
+
+Snapshot File is JSON, it includes:
+
+1. version: Snapshot file version, current is 3.
+2. id: snapshot id, same to file name.
+3. schemaId: the corresponding schema version for this commit.
+4. baseManifestList: a manifest list recording all changes from the previous snapshots.
+5. deltaManifestList: a manifest list recording all new changes occurred in this snapshot.
+6. changelogManifestList: a manifest list recording all changelog produced in this snapshot, null if no changelog is produced.
+7. indexManifest: a manifest recording all index files of this table, null if no index file.
+8. commitUser: usually generated by UUID, it is used for recovery of streaming writes, one stream write job with one user.
+9. commitIdentifier: transaction id corresponding to streaming write, each transaction may result in multiple commits for different commitKinds.
+10. commitKind: type of changes in this snapshot, including append, compact, overwrite and analyze.
+11. timeMillis: commit time millis.
+12. totalRecordCount: record count of all changes occurred in this snapshot.
+13. deltaRecordCount: record count of all new changes occurred in this snapshot.
+14. changelogRecordCount: record count of all changelog produced in this snapshot.
+15. watermark: watermark for input records, from Flink watermark mechanism, null if there is no watermark.
+16. statistics: stats file name for statistics of this table.
+
+## Manifest List
+
+Manifest List includes meta of several manifest files. Its name contains UUID, it is a avro file, the schema is:
+
+1. fileName: manifest file name.
+2. fileSize: manifest file size.
+3. numAddedFiles: number added files in manifest.
+4. numDeletedFiles: number deleted files in manifest.
+5. partitionStats: partition stats, the minimum and maximum values of partition fields in this manifest are beneficial
+ for skipping certain manifest files during queries, it is a SimpleStats.
+6. schemaId: schema id when writing this manifest file.
+
+## Manifest
+
+Manifest includes meta of several data files or changelog files. Its name contains UUID, it is a avro file.
+
+The changes of the file are saved in the manifest, and the file can be added or deleted. Manifests should be in
+an orderly manner, and the same file may be added or deleted multiple times. The last version should be read. This
+design can make commit lighter to support file deletion generated by compaction.
+
+The schema is:
+
+1. kind: ADD or DELETE,
+2. partition: partition spec, a BinaryRow.
+3. bucket: bucket of this file.
+4. totalBuckets: total buckets when write this file, it is used for verification after bucket changes.
+5. file: data file meta.
+
+The data file meta is:
+
+1. fileName: file name.
+2. fileSize: file size.
+3. rowCount: total number of rows (including add & delete) in this file.
+4. minKey: the minimum key of this file.
+5. minKey: the maximum key of this file.
+6. keyStats: the statistics of the key.
+7. valueStats: the statistics of the value.
+8. minSequenceNumber: the minimum sequence number.
+9. maxSequenceNumber: the maximum sequence number.
+10. schemaId: schema id when write this file.
+11. level: level of this file, in LSM.
+12. extraFiles: extra files for this file, for example, data file index file.
+13. creationTime: creation time of this file.
+14. deleteRowCount: rowCount = addRowCount + deleteRowCount.
+15. embeddedIndex: if data file index is too small, store the index in manifest.
+
+## Partition
+
+Consider a Partition table via Flink SQL:
+
+```sql
+CREATE TABLE part_t (
+ f0 INT,
+ f1 STRING,
+ dt STRING
+) PARTITIONED BY (dt);
+
+INSERT INTO part_t VALUES (1, '11', '20240514');
+```
+
+The file system will be:
+
+```shell
+part_t
+├── dt=20240514
+│ └── bucket-0
+│ └── data-ca1c3c38-dc8d-4533-949b-82e195b41bd4-0.orc
+├── manifest
+│ ├── manifest-08995fe5-c2ac-4f54-9a5f-d3af1fcde41d-0
+│ ├── manifest-list-51c16f7b-421c-4bc0-80a0-17677f343358-0
+│ └── manifest-list-51c16f7b-421c-4bc0-80a0-17677f343358-1
+├── schema
+│ └── schema-0
+└── snapshot
+ ├── EARLIEST
+ ├── LATEST
+ └── snapshot-1
+```
+
+Paimon adopts the same partitioning concept as Apache Hive to separate data. The files of the partition will be placed
+in a separate partition directory.
+
+## Bucket
+
+The storage of all Paimon tables relies on buckets, and data files are stored in the bucket directory. The
+relationship between various table types and buckets in Paimon:
+
+1. Primary Key Table:
+ 1. bucket = -1: Default mode, the dynamic bucket mode records which bucket the key corresponds to through the index
+ files. The index records the correspondence between the hash value of the primary-key and the bucket.
+ 2. bucket = 10: The data is distributed to the corresponding buckets according to the hash value of bucket key (
+ default is primary key).
+2. Append Table:
+ 1. bucket = -1: Default mode, ignoring bucket concept, although all data is written to bucket-0, the parallelism of
+ reads and writes is unrestricted.
+ 2. bucket = 10: You need to define bucket-key too, the data is distributed to the corresponding buckets according to
+ the hash value of bucket key.
+
+## Data File
+
+The name of data file is `data-${uuid}-${id}.${format}`. For the append table, the file stores the data of the table
+without adding any new columns. But for the primary key table, each row of data stores additional system columns:
+
+1. `_VALUE_KIND`: row is deleted or added. Similar to RocksDB, each row of data can be deleted or added, which will be
+ used for updating the primary key table.
+2. `_SEQUENCE_NUMBER`: this number is used for comparison during updates, determining which data came first and which
+ data came later.
+3. `_KEY_` prefix to key columns, this is to avoid conflicts with columns of the table.
+
+## Changelog File
+
+Changelog file and Data file are exactly the same, it only takes effect on the primary key table. It is similar to the
+Binlog in a database, recording changes to the data in the table.
+
+## Global Index
+
+Global Index is in the index directory, currently, only two places will use global index:
+
+1. bucket = -1 + primary key table: in dynamic bucket mode, the index records the correspondence between the hash value
+ of the primary-key and the bucket, each bucket has an index file.
+2. Deletion Vectors: index stores the deletion file, and each bucket has a deletion file.
+
+## Data File Index
+
+Define `file-index.bloom-filter.columns`, Paimon will create its corresponding index file for each file. If the index
+file is too small, it will be stored directly in the manifest, or in the directory of the data file. Each data file
+corresponds to an index file, which has a separate file definition and can contain different types of indexes with
+multiple columns.
(paimon) 19/31: [spark][core] fix when a table have no snapshot or no statistics (#3341)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 9e63ea6e2bcf1a1587bc3d8b32cda6e6ed76d03d
Author: Yann Byron <bi...@gmail.com>
AuthorDate: Fri May 17 12:58:53 2024 +0800
[spark][core] fix when a table have no snapshot or no statistics (#3341)
---
.../apache/paimon/table/system/StatisticTable.java | 26 ++++++++++++++--------
.../commands/PaimonAnalyzeTableColumnCommand.scala | 11 ++++-----
.../paimon/spark/sql/AnalyzeTableTestBase.scala | 8 +++++++
3 files changed, 31 insertions(+), 14 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/system/StatisticTable.java b/paimon-core/src/main/java/org/apache/paimon/table/system/StatisticTable.java
index 2f795fd4a..682b5b775 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/system/StatisticTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/system/StatisticTable.java
@@ -25,6 +25,7 @@ import org.apache.paimon.disk.IOManager;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.Path;
import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.reader.EmptyRecordReader;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.stats.Statistics;
import org.apache.paimon.table.FileStoreTable;
@@ -53,6 +54,7 @@ import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.Optional;
import static org.apache.paimon.catalog.Catalog.SYSTEM_TABLE_SPLITTER;
@@ -204,16 +206,22 @@ public class StatisticTable implements ReadonlyTable {
if (!(split instanceof StatisticTable.StatisticSplit)) {
throw new IllegalArgumentException("Unsupported split: " + split.getClass());
}
- Statistics statistics = dataTable.statistics().get();
- Iterator<Statistics> statisticsIterator =
- Collections.singletonList(statistics).iterator();
- Iterator<InternalRow> rows = Iterators.transform(statisticsIterator, this::toRow);
- if (projection != null) {
- rows =
- Iterators.transform(
- rows, row -> ProjectedRow.from(projection).replaceRow(row));
+
+ Optional<Statistics> statisticsOptional = dataTable.statistics();
+ if (statisticsOptional.isPresent()) {
+ Statistics statistics = statisticsOptional.get();
+ Iterator<Statistics> statisticsIterator =
+ Collections.singletonList(statistics).iterator();
+ Iterator<InternalRow> rows = Iterators.transform(statisticsIterator, this::toRow);
+ if (projection != null) {
+ rows =
+ Iterators.transform(
+ rows, row -> ProjectedRow.from(projection).replaceRow(row));
+ }
+ return new IteratorRecordReader<>(rows);
+ } else {
+ return new EmptyRecordReader<>();
}
- return new IteratorRecordReader<>(rows);
}
private InternalRow toRow(Statistics statistics) {
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonAnalyzeTableColumnCommand.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonAnalyzeTableColumnCommand.scala
index 5f9957dbe..b3eb8059f 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonAnalyzeTableColumnCommand.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonAnalyzeTableColumnCommand.scala
@@ -53,10 +53,14 @@ case class PaimonAnalyzeTableColumnCommand(
override def run(sparkSession: SparkSession): Seq[Row] = {
val relation = DataSourceV2Relation.create(v2Table, Some(catalog), Some(identifier))
+ val attributes = getColumnsToAnalyze(relation)
+
val currentSnapshot = table.snapshotManager().latestSnapshot()
+ if (currentSnapshot == null) {
+ return Seq.empty[Row]
+ }
// compute stats
- val attributes = getColumnsToAnalyze(relation, columnNames, allColumns)
val totalSize = PaimonStatsUtils.calculateTotalSize(
sparkSession.sessionState,
table.name(),
@@ -92,10 +96,7 @@ case class PaimonAnalyzeTableColumnCommand(
Seq.empty[Row]
}
- private def getColumnsToAnalyze(
- relation: DataSourceV2Relation,
- columnNames: Option[Seq[String]],
- allColumns: Boolean): Seq[Attribute] = {
+ private def getColumnsToAnalyze(relation: DataSourceV2Relation): Seq[Attribute] = {
if (columnNames.isDefined && allColumns) {
throw new UnsupportedOperationException(
"Parameter `columnNames` and `allColumns` are " +
diff --git a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/AnalyzeTableTestBase.scala b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/AnalyzeTableTestBase.scala
index d1e6f7350..9ff252b52 100644
--- a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/AnalyzeTableTestBase.scala
+++ b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/AnalyzeTableTestBase.scala
@@ -59,6 +59,7 @@ abstract class AnalyzeTableTestBase extends PaimonSparkTestBase {
spark.sql(s"INSERT INTO T VALUES ('1', 'a', 1, 1)")
spark.sql(s"INSERT INTO T VALUES ('2', 'aaa', 1, 2)")
+ Assertions.assertEquals(0, spark.sql("select * from `T$statistics`").count())
spark.sql(s"ANALYZE TABLE T COMPUTE STATISTICS")
@@ -70,6 +71,13 @@ abstract class AnalyzeTableTestBase extends PaimonSparkTestBase {
Row(2, 0, 2, "{ }"))
}
+ test("Paimon analyze: analyze table without snapshot") {
+ spark.sql(s"CREATE TABLE T (id STRING, name STRING)")
+ spark.sql(s"ANALYZE TABLE T COMPUTE STATISTICS")
+ spark.sql(s"ANALYZE TABLE T COMPUTE STATISTICS FOR ALL COLUMNS")
+ Assertions.assertEquals(0, spark.sql("select * from `T$statistics`").count())
+ }
+
test("Paimon analyze: analyze no scan") {
spark.sql(s"CREATE TABLE T (id STRING, name STRING)")
assertThatThrownBy(() => spark.sql(s"ANALYZE TABLE T COMPUTE STATISTICS NOSCAN"))
(paimon) 06/31: [hive] fix db-location conflict with location-in-properties (#3309)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 45be92ac2f236aadac93debc6d6e80c6d59f242b
Author: juncheng yin <40...@users.noreply.github.com>
AuthorDate: Sun May 12 23:00:59 2024 +0800
[hive] fix db-location conflict with location-in-properties (#3309)
---
.../paimon/hive/TBPropertiesLocationHelper.java | 1 +
.../apache/paimon/hive/HiveCatalogITCaseBase.java | 39 ++++++++++++++++++++++
2 files changed, 40 insertions(+)
diff --git a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/TBPropertiesLocationHelper.java b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/TBPropertiesLocationHelper.java
index 8f4ceee6d..c4fe2a94b 100644
--- a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/TBPropertiesLocationHelper.java
+++ b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/TBPropertiesLocationHelper.java
@@ -69,6 +69,7 @@ public final class TBPropertiesLocationHelper implements LocationHelper {
}
properties.put(LocationKeyExtractor.TBPROPERTIES_LOCATION_KEY, path.toString());
database.setParameters(properties);
+ database.setLocationUri(null);
}
@Override
diff --git a/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/HiveCatalogITCaseBase.java b/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/HiveCatalogITCaseBase.java
index a5c494c1e..e478d44cd 100644
--- a/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/HiveCatalogITCaseBase.java
+++ b/paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/HiveCatalogITCaseBase.java
@@ -42,6 +42,7 @@ import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException;
import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException;
import org.apache.flink.types.Row;
import org.apache.flink.util.CloseableIterator;
+import org.apache.hadoop.hive.metastore.api.MetaException;
import org.junit.Rule;
import org.junit.Test;
import org.junit.jupiter.api.function.Executable;
@@ -159,6 +160,44 @@ public abstract class HiveCatalogITCaseBase {
}
};
+ @Test
+ public void testDbLocation() {
+ String dbLocation = minioTestContainer.getS3UriForDefaultBucket() + "/" + UUID.randomUUID();
+ Catalog catalog =
+ ((FlinkCatalog) tEnv.getCatalog(tEnv.getCurrentCatalog()).get()).catalog();
+ Map<String, String> properties = new HashMap<>();
+ properties.put("location", dbLocation);
+
+ assertThatThrownBy(() -> catalog.createDatabase("location_test_db", false, properties))
+ .hasRootCauseInstanceOf(MetaException.class)
+ .hasRootCauseMessage(
+ "Got exception: java.io.IOException No FileSystem for scheme: s3");
+ }
+
+ @Test
+ @LocationInProperties
+ public void testDbLocationWithMetastoreLocationInProperties()
+ throws Catalog.DatabaseAlreadyExistException {
+ String dbLocation = minioTestContainer.getS3UriForDefaultBucket() + "/" + UUID.randomUUID();
+ Catalog catalog =
+ ((FlinkCatalog) tEnv.getCatalog(tEnv.getCurrentCatalog()).get()).catalog();
+ Map<String, String> properties = new HashMap<>();
+ properties.put("location", dbLocation);
+
+ catalog.createDatabase("location_test_db", false, properties);
+ assertThat(catalog.databaseExists("location_test_db"));
+
+ hiveShell.execute("USE location_test_db");
+ hiveShell.execute("CREATE TABLE location_test_db ( a INT, b INT )");
+ hiveShell.execute("INSERT INTO location_test_db VALUES (1, 100)");
+ hiveShell.execute("INSERT INTO location_test_db VALUES (2, 200)");
+
+ assertThat(hiveShell.executeQuery("SELECT * from location_test_db"))
+ .containsExactlyInAnyOrder("1\t100", "2\t200");
+
+ hiveShell.execute("DROP DATABASE IF EXISTS location_test_db CASCADE");
+ }
+
@Test
public void testDatabaseOperations() throws Exception {
// create database
(paimon) 11/31: [cdc] Add more checker for the timestamp type. (#3312)
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 636b9fa58af76f0e2ba704f35d89cc727cb80344
Author: MOBIN <18...@163.com>
AuthorDate: Mon May 13 16:29:44 2024 +0800
[cdc] Add more checker for the timestamp type. (#3312)
---
.../cdc/UpdatedDataFieldsProcessFunctionBase.java | 11 +++++++++++
.../cdc/UpdatedDataFieldsProcessFunctionBaseTest.java | 19 +++++++++++++++++++
2 files changed, 30 insertions(+)
diff --git a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/UpdatedDataFieldsProcessFunctionBase.java b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/UpdatedDataFieldsProcessFunctionBase.java
index bc31a05e2..8fefcab63 100644
--- a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/UpdatedDataFieldsProcessFunctionBase.java
+++ b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/UpdatedDataFieldsProcessFunctionBase.java
@@ -63,6 +63,9 @@ public abstract class UpdatedDataFieldsProcessFunctionBase<I, O> extends Process
private static final List<DataTypeRoot> DECIMAL_TYPES = Arrays.asList(DataTypeRoot.DECIMAL);
+ private static final List<DataTypeRoot> TIMESTAMP_TYPES =
+ Arrays.asList(DataTypeRoot.TIMESTAMP_WITHOUT_TIME_ZONE);
+
protected UpdatedDataFieldsProcessFunctionBase(Catalog.Loader catalogLoader) {
this.catalogLoader = catalogLoader;
}
@@ -174,6 +177,14 @@ public abstract class UpdatedDataFieldsProcessFunctionBase<I, O> extends Process
: ConvertAction.CONVERT;
}
+ oldIdx = TIMESTAMP_TYPES.indexOf(oldType.getTypeRoot());
+ newIdx = TIMESTAMP_TYPES.indexOf(newType.getTypeRoot());
+ if (oldIdx >= 0 && newIdx >= 0) {
+ return DataTypeChecks.getPrecision(oldType) <= DataTypeChecks.getPrecision(newType)
+ ? ConvertAction.CONVERT
+ : ConvertAction.IGNORE;
+ }
+
return ConvertAction.EXCEPTION;
}
diff --git a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/UpdatedDataFieldsProcessFunctionBaseTest.java b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/UpdatedDataFieldsProcessFunctionBaseTest.java
index 6b1ed71ec..08d9ac975 100644
--- a/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/UpdatedDataFieldsProcessFunctionBaseTest.java
+++ b/paimon-flink/paimon-flink-cdc/src/test/java/org/apache/paimon/flink/sink/cdc/UpdatedDataFieldsProcessFunctionBaseTest.java
@@ -22,6 +22,7 @@ import org.apache.paimon.types.BigIntType;
import org.apache.paimon.types.DecimalType;
import org.apache.paimon.types.IntType;
import org.apache.paimon.types.SmallIntType;
+import org.apache.paimon.types.TimestampType;
import org.apache.paimon.types.VarCharType;
import org.junit.Assert;
@@ -79,4 +80,22 @@ public class UpdatedDataFieldsProcessFunctionBaseTest {
Assert.assertEquals(
UpdatedDataFieldsProcessFunctionBase.ConvertAction.IGNORE, convertAction);
}
+
+ @Test
+ public void testCanConvertTimestamp() {
+ TimestampType oldType = new TimestampType(true, 3);
+ TimestampType biggerLengthTimestamp = new TimestampType(true, 5);
+ TimestampType smallerLengthTimestamp = new TimestampType(true, 2);
+
+ UpdatedDataFieldsProcessFunctionBase.ConvertAction convertAction = null;
+ convertAction =
+ UpdatedDataFieldsProcessFunctionBase.canConvert(oldType, biggerLengthTimestamp);
+ Assert.assertEquals(
+ UpdatedDataFieldsProcessFunctionBase.ConvertAction.CONVERT, convertAction);
+ convertAction =
+ UpdatedDataFieldsProcessFunctionBase.canConvert(oldType, smallerLengthTimestamp);
+
+ Assert.assertEquals(
+ UpdatedDataFieldsProcessFunctionBase.ConvertAction.IGNORE, convertAction);
+ }
}
(paimon) 01/31: [minor] Add Nullable to logOffsets in Snapshot
Posted by lz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch release-0.8
in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 499e9c5db4f284043fd52ddfc5eaf18f6896b32a
Author: Jingsong <ji...@gmail.com>
AuthorDate: Mon May 6 15:12:55 2024 +0800
[minor] Add Nullable to logOffsets in Snapshot
---
paimon-core/src/main/java/org/apache/paimon/Snapshot.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/Snapshot.java b/paimon-core/src/main/java/org/apache/paimon/Snapshot.java
index fefc41058..20d414092 100644
--- a/paimon-core/src/main/java/org/apache/paimon/Snapshot.java
+++ b/paimon-core/src/main/java/org/apache/paimon/Snapshot.java
@@ -232,7 +232,7 @@ public class Snapshot {
@JsonProperty(FIELD_COMMIT_IDENTIFIER) long commitIdentifier,
@JsonProperty(FIELD_COMMIT_KIND) CommitKind commitKind,
@JsonProperty(FIELD_TIME_MILLIS) long timeMillis,
- @JsonProperty(FIELD_LOG_OFFSETS) Map<Integer, Long> logOffsets,
+ @JsonProperty(FIELD_LOG_OFFSETS) @Nullable Map<Integer, Long> logOffsets,
@JsonProperty(FIELD_TOTAL_RECORD_COUNT) @Nullable Long totalRecordCount,
@JsonProperty(FIELD_DELTA_RECORD_COUNT) @Nullable Long deltaRecordCount,
@JsonProperty(FIELD_CHANGELOG_RECORD_COUNT) @Nullable Long changelogRecordCount,