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 2023/09/06 03:20:24 UTC
[incubator-paimon] branch release-0.5 updated (a4d7ad054 -> 6f5fa5727)
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a change to branch release-0.5
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
from a4d7ad054 [doc] Update stable 0.5
new f03a5bee1 [doc] Document more hadoop-conf-dir
new b974661db [doc] Update confused documentation
new 23280843f [doc] Update trino doc to trino 422 (#1884)
new e6a98ba95 [doc] Fix minor SQL error in trino.md (#1882)
new 461096ef6 [doc] Document Spark truncate table
new 4345fa60c [doc] Document performance for dynamic bucket
new 10d933cfb [doc] Document warehouse dir for FlinkGenericCatalog
new cb4dbc28a [doc] Correct the errors in the document (#1928)
new 52be707ed [flink] Fix global dynamic bucket mode (#1914)
new d7b8107ed [doc] Add presto sql 332 to download
new 6f5fa5727 [doc] Document Cross Partitions Update Dynamic Bucket Mode not work in 0.5.0
The 11 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:
docs/content/api/java-api.md | 1 +
docs/content/concepts/primary-key-table.md | 19 +-
docs/content/engines/flink.md | 5 +
docs/content/engines/hive.md | 1 +
docs/content/engines/overview.md | 2 +-
docs/content/engines/presto.md | 2 +
docs/content/engines/trino.md | 36 +-
docs/content/how-to/cdc-ingestion.md | 9 +-
docs/content/how-to/creating-catalogs.md | 5 +-
docs/content/how-to/writing-tables.md | 17 +-
docs/content/project/download.md | 2 +
.../data/serializer/InternalRowSerializer.java | 2 +-
.../paimon/disk/BufferFileReaderInputView.java | 99 ++++++
.../org/apache/paimon/disk/ExternalBuffer.java | 383 +++++++++++++++++++++
.../org/apache/paimon/disk/ExternalBufferTest.java | 209 +++++++++++
.../paimon/flink/sink/cdc/CdcSinkBuilder.java | 4 -
.../sink/cdc/FlinkCdcSyncDatabaseSinkBuilder.java | 3 -
.../flink/sink/index/GlobalDynamicBucketSink.java | 16 +-
.../sink/index/GlobalDynamicCdcBucketSink.java | 133 -------
.../flink/sink/index/GlobalIndexAssigner.java | 29 +-
.../sink/index/GlobalIndexAssignerOperator.java | 107 ++++--
.../paimon/flink/sink/index/IndexBootstrap.java | 48 ++-
.../sink/index/KeyPartRowChannelComputer.java | 6 +-
.../flink/GlobalDynamicBucketTableITCase.java | 19 +
.../sink/cdc/FlinkCdcSyncTableSinkITCase.java | 2 +
.../flink/sink/index/IndexBootstrapTest.java | 17 +-
26 files changed, 942 insertions(+), 234 deletions(-)
create mode 100644 paimon-core/src/main/java/org/apache/paimon/disk/BufferFileReaderInputView.java
create mode 100644 paimon-core/src/main/java/org/apache/paimon/disk/ExternalBuffer.java
create mode 100644 paimon-core/src/test/java/org/apache/paimon/disk/ExternalBufferTest.java
delete mode 100644 paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/index/GlobalDynamicCdcBucketSink.java
[incubator-paimon] 06/11: [doc] Document performance for dynamic bucket
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.5
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
commit 4345fa60c45ddc9ee12fd3dd4008e5553f1b31b3
Author: Jingsong <ji...@gmail.com>
AuthorDate: Mon Aug 28 19:33:00 2023 +0800
[doc] Document performance for dynamic bucket
---
docs/content/concepts/primary-key-table.md | 17 +++++++++++++----
1 file changed, 13 insertions(+), 4 deletions(-)
diff --git a/docs/content/concepts/primary-key-table.md b/docs/content/concepts/primary-key-table.md
index d82f3d6e3..982af7a30 100644
--- a/docs/content/concepts/primary-key-table.md
+++ b/docs/content/concepts/primary-key-table.md
@@ -53,13 +53,18 @@ Configure `'bucket' = '-1'`, Paimon dynamically maintains the index, automatic e
Dynamic Bucket only support single write job. Please do not start multiple jobs to write to the same partition.
{{< /hint >}}
-**Normal Dynamic Bucket Mode**:
+#### Normal Dynamic Bucket Mode
When your updates do not cross partitions (no partitions, or primary keys contain all partition fields), Dynamic
-Bucket mode uses HASH index to maintain mapping from key to bucket, it requires more memory than fixed bucket mode,
-100 million entries in a partition takes up 1 GB more memory, partitions that are no longer active do not take up memory.
+Bucket mode uses HASH index to maintain mapping from key to bucket, it requires more memory than fixed bucket mode.
-**Cross Partitions Update Dynamic Bucket Mode**:
+Performance:
+
+1. Generally speaking, there is no performance loss, but there will be some additional memory consumption, **100 million**
+ entries in a partition takes up **1 GB** more memory, partitions that are no longer active do not take up memory.
+2. For tables with low update rates, this mode is recommended to significantly improve performance.
+
+#### Cross Partitions Update Dynamic Bucket Mode
{{< hint info >}}
This is an experimental feature.
@@ -68,10 +73,14 @@ This is an experimental feature.
When you need cross partition updates (primary keys not contain all partition fields), Dynamic Bucket mode directly
maintains the mapping of keys to partition and bucket, uses local disks, and initializes indexes by reading all
existing keys in the table when starting stream write job. Different merge engines have different behaviors:
+
1. Deduplicate: Delete data from the old partition and insert new data into the new partition.
2. PartialUpdate & Aggregation: Insert new data into the old partition.
3. FirstRow: Ignore new data if there is old value.
+Performance: For tables with a large amount of data, there will be a significant loss in performance. Moreover,
+initialization takes a long time.
+
## Merge Engines
When Paimon sink receives two or more records with the same primary keys, it will merge them into one record to keep primary keys unique. By specifying the `merge-engine` table property, users can choose how records are merged together.
[incubator-paimon] 08/11: [doc] Correct the errors in the document (#1928)
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.5
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
commit cb4dbc28ab1ad337511255cc7518cdc2dea838a3
Author: monster <60...@users.noreply.github.com>
AuthorDate: Thu Aug 31 18:26:22 2023 +0800
[doc] Correct the errors in the document (#1928)
---
docs/content/how-to/cdc-ingestion.md | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/docs/content/how-to/cdc-ingestion.md b/docs/content/how-to/cdc-ingestion.md
index b7158e335..45234de23 100644
--- a/docs/content/how-to/cdc-ingestion.md
+++ b/docs/content/how-to/cdc-ingestion.md
@@ -274,7 +274,7 @@ flink-sql-connector-kafka-*.jar
```
### Supported Formats
-Flink provides several Kafka CDC formats :canal-json、debezium-json,ogg-json,maxwell-json.
+Flink provides several Kafka CDC formats: Canal, Debezium, Ogg and Maxwell JSON.
If a message in a Kafka topic is a change event captured from another database using the Change Data Capture (CDC) tool, then you can use the Paimon Kafka CDC. Write the INSERT, UPDATE, DELETE messages parsed into the paimon table.
<table class="table table-bordered">
<thead>
[incubator-paimon] 11/11: [doc] Document Cross Partitions Update Dynamic Bucket Mode not work in 0.5.0
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.5
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
commit 6f5fa5727ee814f7030d2e88a692a8aa280e786c
Author: Jingsong <ji...@gmail.com>
AuthorDate: Wed Sep 6 11:17:34 2023 +0800
[doc] Document Cross Partitions Update Dynamic Bucket Mode not work in 0.5.0
---
docs/content/concepts/primary-key-table.md | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/docs/content/concepts/primary-key-table.md b/docs/content/concepts/primary-key-table.md
index 982af7a30..feb8c20ff 100644
--- a/docs/content/concepts/primary-key-table.md
+++ b/docs/content/concepts/primary-key-table.md
@@ -67,7 +67,7 @@ Performance:
#### Cross Partitions Update Dynamic Bucket Mode
{{< hint info >}}
-This is an experimental feature.
+This is an experimental feature. Release `0.5.0-incubating` is not available, please use a higher version.
{{< /hint >}}
When you need cross partition updates (primary keys not contain all partition fields), Dynamic Bucket mode directly
[incubator-paimon] 01/11: [doc] Document more hadoop-conf-dir
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.5
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
commit f03a5bee17f9b1993314fb37b4746a362980a782
Author: Jingsong <ji...@gmail.com>
AuthorDate: Thu Aug 24 23:02:52 2023 +0800
[doc] Document more hadoop-conf-dir
---
docs/content/api/java-api.md | 1 +
docs/content/engines/hive.md | 1 +
docs/content/how-to/creating-catalogs.md | 5 ++++-
3 files changed, 6 insertions(+), 1 deletion(-)
diff --git a/docs/content/api/java-api.md b/docs/content/api/java-api.md
index 9ddfcc0f4..ed72d6b01 100644
--- a/docs/content/api/java-api.md
+++ b/docs/content/api/java-api.md
@@ -70,6 +70,7 @@ public class CreateCatalog {
options.set("metastore", "hive");
options.set("uri", "...");
options.set("hive-conf-dir", "...");
+ options.set("hadoop-conf-dir", "...");
CatalogContext context = CatalogContext.create(options);
Catalog catalog = CatalogFactory.createCatalog(context);
}
diff --git a/docs/content/engines/hive.md b/docs/content/engines/hive.md
index 16866e06e..2de5aee30 100644
--- a/docs/content/engines/hive.md
+++ b/docs/content/engines/hive.md
@@ -102,6 +102,7 @@ CREATE CATALOG my_hive WITH (
'metastore' = 'hive',
'uri' = 'thrift://<hive-metastore-host-name>:<port>',
-- 'hive-conf-dir' = '...', this is recommended in the kerberos environment
+ -- 'hadoop-conf-dir' = '...', this is recommended in the kerberos environment
'warehouse' = 'hdfs:///path/to/table/store/warehouse'
);
diff --git a/docs/content/how-to/creating-catalogs.md b/docs/content/how-to/creating-catalogs.md
index d76832caa..11f2d7925 100644
--- a/docs/content/how-to/creating-catalogs.md
+++ b/docs/content/how-to/creating-catalogs.md
@@ -90,7 +90,9 @@ Paimon Hive catalog in Flink relies on Flink Hive connector bundled jar. You sho
The following Flink SQL registers and uses a Paimon Hive catalog named `my_hive`. Metadata and table files are stored under `hdfs:///path/to/warehouse`. In addition, metadata is also stored in Hive metastore.
-If your Hive requires security authentication such as Kerberos, LDAP, Ranger or you want the paimon table to be managed by Apache Atlas(Setting 'hive.metastore.event.listeners' in hive-site.xml). You can specify the hive-conf-dir parameter to the hive-site.xml file path.
+If your Hive requires security authentication such as Kerberos, LDAP, Ranger or you want the paimon table to be managed
+by Apache Atlas(Setting 'hive.metastore.event.listeners' in hive-site.xml). You can specify the hive-conf-dir and
+hadoop-conf-dir parameter to the hive-site.xml file path.
```sql
CREATE CATALOG my_hive WITH (
@@ -98,6 +100,7 @@ CREATE CATALOG my_hive WITH (
'metastore' = 'hive',
'uri' = 'thrift://<hive-metastore-host-name>:<port>',
-- 'hive-conf-dir' = '...', this is recommended in the kerberos environment
+ -- 'hadoop-conf-dir' = '...', this is recommended in the kerberos environment
'warehouse' = 'hdfs:///path/to/warehouse'
);
[incubator-paimon] 02/11: [doc] Update confused documentation
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.5
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
commit b974661dbe63baeaf55290ea9d788c5dab3e2e98
Author: Jingsong <ji...@gmail.com>
AuthorDate: Fri Aug 25 14:56:58 2023 +0800
[doc] Update confused documentation
---
docs/content/how-to/cdc-ingestion.md | 7 ++++---
1 file changed, 4 insertions(+), 3 deletions(-)
diff --git a/docs/content/how-to/cdc-ingestion.md b/docs/content/how-to/cdc-ingestion.md
index e753542ee..b7158e335 100644
--- a/docs/content/how-to/cdc-ingestion.md
+++ b/docs/content/how-to/cdc-ingestion.md
@@ -603,10 +603,11 @@ behaviors of `RENAME TABLE` and `DROP COLUMN` will be ignored, `RENAME COLUMN` w
{{< generated/compute_column >}}
## Special Data Type Mapping
+
1. MySQL TINYINT(1) type will be mapped to Boolean by default. If you want to store number (-128~127) in it like MySQL,
-you can specify type mapping option `tinyint1-not-bool`, then the column will be mapped to TINYINT in Paimon table.
-2. You can use type mapping option `to-nullable` to ignore all NOT NULL constraints (except primary keys).
-3. You can use type mapping option `to-string` to map all MySQL data type to STRING.
+you can specify type mapping option `tinyint1-not-bool` (Use `--type-mapping`), then the column will be mapped to TINYINT in Paimon table.
+2. You can use type mapping option `to-nullable` (Use `--type-mapping`) to ignore all NOT NULL constraints (except primary keys).
+3. You can use type mapping option `to-string` (Use `--type-mapping`) to map all MySQL data type to STRING.
4. MySQL BIT(1) type will be mapped to Boolean.
5. When using Hive catalog, MySQL TIME type will be mapped to STRING.
6. MySQL BINARY will be mapped to Paimon VARBINARY. This is because the binary value is passed as bytes in binlog, so it
[incubator-paimon] 09/11: [flink] Fix global dynamic bucket mode (#1914)
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.5
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
commit 52be707ed3e1bf38291b2539744a5dab8fc2fbc0
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Thu Aug 31 19:07:37 2023 +0800
[flink] Fix global dynamic bucket mode (#1914)
---
.../data/serializer/InternalRowSerializer.java | 2 +-
.../paimon/disk/BufferFileReaderInputView.java | 99 ++++++
.../org/apache/paimon/disk/ExternalBuffer.java | 383 +++++++++++++++++++++
.../org/apache/paimon/disk/ExternalBufferTest.java | 209 +++++++++++
.../paimon/flink/sink/cdc/CdcSinkBuilder.java | 4 -
.../sink/cdc/FlinkCdcSyncDatabaseSinkBuilder.java | 3 -
.../flink/sink/index/GlobalDynamicBucketSink.java | 16 +-
.../sink/index/GlobalDynamicCdcBucketSink.java | 133 -------
.../flink/sink/index/GlobalIndexAssigner.java | 29 +-
.../sink/index/GlobalIndexAssignerOperator.java | 107 ++++--
.../paimon/flink/sink/index/IndexBootstrap.java | 48 ++-
.../sink/index/KeyPartRowChannelComputer.java | 6 +-
.../flink/GlobalDynamicBucketTableITCase.java | 19 +
.../sink/cdc/FlinkCdcSyncTableSinkITCase.java | 2 +
.../flink/sink/index/IndexBootstrapTest.java | 17 +-
15 files changed, 875 insertions(+), 202 deletions(-)
diff --git a/paimon-common/src/main/java/org/apache/paimon/data/serializer/InternalRowSerializer.java b/paimon-common/src/main/java/org/apache/paimon/data/serializer/InternalRowSerializer.java
index b0701be48..0f573f10b 100644
--- a/paimon-common/src/main/java/org/apache/paimon/data/serializer/InternalRowSerializer.java
+++ b/paimon-common/src/main/java/org/apache/paimon/data/serializer/InternalRowSerializer.java
@@ -72,7 +72,7 @@ public class InternalRowSerializer extends AbstractRowDataSerializer<InternalRow
}
@Override
- public Serializer<InternalRow> duplicate() {
+ public InternalRowSerializer duplicate() {
Serializer<?>[] duplicateFieldSerializers = new Serializer[fieldSerializers.length];
for (int i = 0; i < fieldSerializers.length; i++) {
duplicateFieldSerializers[i] = fieldSerializers[i].duplicate();
diff --git a/paimon-core/src/main/java/org/apache/paimon/disk/BufferFileReaderInputView.java b/paimon-core/src/main/java/org/apache/paimon/disk/BufferFileReaderInputView.java
new file mode 100644
index 000000000..ab91b8944
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/disk/BufferFileReaderInputView.java
@@ -0,0 +1,99 @@
+/*
+ * 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.disk;
+
+import org.apache.paimon.data.AbstractPagedInputView;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.serializer.BinaryRowSerializer;
+import org.apache.paimon.memory.Buffer;
+import org.apache.paimon.memory.MemorySegment;
+import org.apache.paimon.utils.MutableObjectIterator;
+
+import java.io.EOFException;
+import java.io.IOException;
+
+/** An {@link AbstractPagedInputView} which reads blocks from channel without compression. */
+public class BufferFileReaderInputView extends AbstractPagedInputView {
+
+ private final BufferFileReader reader;
+ private final MemorySegment segment;
+
+ private int currentSegmentLimit;
+
+ public BufferFileReaderInputView(FileIOChannel.ID id, IOManager ioManager, int segmentSize)
+ throws IOException {
+ this.reader = ioManager.createBufferFileReader(id);
+ this.segment = MemorySegment.wrap(new byte[segmentSize]);
+ }
+
+ @Override
+ protected MemorySegment nextSegment(MemorySegment current) throws IOException {
+ if (reader.hasReachedEndOfFile()) {
+ throw new EOFException();
+ }
+
+ Buffer buffer = Buffer.create(segment);
+ reader.readInto(buffer);
+ this.currentSegmentLimit = buffer.getSize();
+ return segment;
+ }
+
+ @Override
+ protected int getLimitForSegment(MemorySegment segment) {
+ return currentSegmentLimit;
+ }
+
+ public void close() throws IOException {
+ reader.close();
+ }
+
+ public FileIOChannel getChannel() {
+ return reader;
+ }
+
+ public MutableObjectIterator<BinaryRow> createBinaryRowIterator(
+ BinaryRowSerializer serializer) {
+ return new BinaryRowChannelInputViewIterator(serializer);
+ }
+
+ private class BinaryRowChannelInputViewIterator implements MutableObjectIterator<BinaryRow> {
+
+ protected final BinaryRowSerializer serializer;
+
+ public BinaryRowChannelInputViewIterator(BinaryRowSerializer serializer) {
+ this.serializer = serializer;
+ }
+
+ @Override
+ public BinaryRow next(BinaryRow reuse) throws IOException {
+ try {
+ return this.serializer.deserializeFromPages(reuse, BufferFileReaderInputView.this);
+ } catch (EOFException e) {
+ close();
+ return null;
+ }
+ }
+
+ @Override
+ public BinaryRow next() throws IOException {
+ throw new UnsupportedOperationException(
+ "This method is disabled due to performance issue!");
+ }
+ }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/disk/ExternalBuffer.java b/paimon-core/src/main/java/org/apache/paimon/disk/ExternalBuffer.java
new file mode 100644
index 000000000..9ced8e83d
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/disk/ExternalBuffer.java
@@ -0,0 +1,383 @@
+/*
+ * 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.disk;
+
+import org.apache.paimon.annotation.VisibleForTesting;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.RandomAccessInputView;
+import org.apache.paimon.data.SimpleCollectingOutputView;
+import org.apache.paimon.data.serializer.AbstractRowDataSerializer;
+import org.apache.paimon.data.serializer.BinaryRowSerializer;
+import org.apache.paimon.memory.Buffer;
+import org.apache.paimon.memory.MemorySegment;
+import org.apache.paimon.memory.MemorySegmentPool;
+import org.apache.paimon.utils.MutableObjectIterator;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.EOFException;
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.paimon.utils.Preconditions.checkState;
+
+/** An external buffer for storing rows, it will spill the data to disk when the memory is full. */
+public class ExternalBuffer {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ExternalBuffer.class);
+
+ private final IOManager ioManager;
+ private final MemorySegmentPool pool;
+ private final BinaryRowSerializer binaryRowSerializer;
+ private final InMemoryBuffer inMemoryBuffer;
+
+ // The size of each segment
+ private final int segmentSize;
+
+ private final List<ChannelWithMeta> spilledChannelIDs;
+ private int numRows;
+
+ private boolean addCompleted;
+
+ public ExternalBuffer(
+ IOManager ioManager, MemorySegmentPool pool, AbstractRowDataSerializer<?> serializer) {
+ this.ioManager = ioManager;
+ this.pool = pool;
+
+ this.binaryRowSerializer =
+ serializer instanceof BinaryRowSerializer
+ ? (BinaryRowSerializer) serializer.duplicate()
+ : new BinaryRowSerializer(serializer.getArity());
+
+ this.segmentSize = pool.pageSize();
+
+ this.spilledChannelIDs = new ArrayList<>();
+
+ this.numRows = 0;
+
+ this.addCompleted = false;
+
+ //noinspection unchecked
+ this.inMemoryBuffer =
+ new InMemoryBuffer((AbstractRowDataSerializer<InternalRow>) serializer);
+ }
+
+ public void reset() {
+ clearChannels();
+ inMemoryBuffer.reset();
+ numRows = 0;
+ addCompleted = false;
+ }
+
+ public void add(InternalRow row) throws IOException {
+ checkState(!addCompleted, "This buffer has add completed.");
+ if (!inMemoryBuffer.write(row)) {
+ // Check if record is too big.
+ if (inMemoryBuffer.getCurrentDataBufferOffset() == 0) {
+ throwTooBigException(row);
+ }
+ spill();
+ if (!inMemoryBuffer.write(row)) {
+ throwTooBigException(row);
+ }
+ }
+
+ numRows++;
+ }
+
+ public void complete() {
+ addCompleted = true;
+ }
+
+ public BufferIterator newIterator() {
+ checkState(addCompleted, "This buffer has not add completed.");
+ return new BufferIterator();
+ }
+
+ private void throwTooBigException(InternalRow row) throws IOException {
+ int rowSize = inMemoryBuffer.serializer.toBinaryRow(row).toBytes().length;
+ throw new IOException(
+ "Record is too big, it can't be added to a empty InMemoryBuffer! "
+ + "Record size: "
+ + rowSize
+ + ", Buffer: "
+ + memorySize());
+ }
+
+ private void spill() throws IOException {
+ FileIOChannel.ID channel = ioManager.createChannel();
+
+ BufferFileWriter writer = ioManager.createBufferFileWriter(channel);
+ int numRecordBuffers = inMemoryBuffer.getNumRecordBuffers();
+ ArrayList<MemorySegment> segments = inMemoryBuffer.getRecordBufferSegments();
+ try {
+ // spill in memory buffer in zero-copy.
+ for (int i = 0; i < numRecordBuffers; i++) {
+ MemorySegment segment = segments.get(i);
+ int bufferSize =
+ i == numRecordBuffers - 1
+ ? inMemoryBuffer.getNumBytesInLastBuffer()
+ : segment.size();
+ writer.writeBlock(Buffer.create(segment, bufferSize));
+ }
+ LOG.info(
+ "here spill the reset buffer data with {} records {} bytes",
+ inMemoryBuffer.numRecords,
+ writer.getSize());
+ writer.close();
+ } catch (IOException e) {
+ writer.closeAndDelete();
+ throw e;
+ }
+
+ spilledChannelIDs.add(
+ new ChannelWithMeta(
+ channel,
+ inMemoryBuffer.getNumRecordBuffers(),
+ inMemoryBuffer.getNumBytesInLastBuffer()));
+
+ inMemoryBuffer.reset();
+ }
+
+ public int size() {
+ return numRows;
+ }
+
+ private int memorySize() {
+ return pool.freePages() * segmentSize;
+ }
+
+ @SuppressWarnings("ResultOfMethodCallIgnored")
+ private void clearChannels() {
+ for (ChannelWithMeta meta : spilledChannelIDs) {
+ final File f = new File(meta.getChannel().getPath());
+ if (f.exists()) {
+ f.delete();
+ }
+ }
+ spilledChannelIDs.clear();
+ }
+
+ /** Iterator of external buffer. */
+ public class BufferIterator implements Closeable {
+
+ private MutableObjectIterator<BinaryRow> currentIterator;
+ private final BinaryRow reuse = binaryRowSerializer.createInstance();
+
+ private int currentChannelID = -1;
+ private BinaryRow row;
+ private boolean closed;
+ private BufferFileReaderInputView channelReader;
+
+ private BufferIterator() {
+ this.closed = false;
+ }
+
+ private void checkValidity() {
+ if (closed) {
+ throw new RuntimeException("This iterator is closed!");
+ }
+ }
+
+ @Override
+ public void close() {
+ if (closed) {
+ return;
+ }
+
+ try {
+ closeCurrentFileReader();
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+
+ closed = true;
+ }
+
+ public boolean advanceNext() {
+ checkValidity();
+
+ try {
+ // get from curr iterator or new iterator.
+ while (true) {
+ if (currentIterator != null && (row = currentIterator.next(reuse)) != null) {
+ return true;
+ } else {
+ if (!nextIterator()) {
+ return false;
+ }
+ }
+ }
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private boolean nextIterator() throws IOException {
+ if (currentChannelID == Integer.MAX_VALUE) {
+ return false;
+ } else if (currentChannelID < spilledChannelIDs.size() - 1) {
+ nextSpilledIterator();
+ } else {
+ newMemoryIterator();
+ }
+ return true;
+ }
+
+ public BinaryRow getRow() {
+ return row;
+ }
+
+ private void closeCurrentFileReader() throws IOException {
+ if (channelReader != null) {
+ channelReader.close();
+ channelReader = null;
+ }
+ }
+
+ private void nextSpilledIterator() throws IOException {
+ ChannelWithMeta channel = spilledChannelIDs.get(currentChannelID + 1);
+ currentChannelID++;
+
+ // close current reader first.
+ closeCurrentFileReader();
+
+ // new reader.
+ this.channelReader =
+ new BufferFileReaderInputView(channel.getChannel(), ioManager, segmentSize);
+ this.currentIterator = channelReader.createBinaryRowIterator(binaryRowSerializer);
+ }
+
+ private void newMemoryIterator() {
+ this.currentChannelID = Integer.MAX_VALUE;
+ this.currentIterator = inMemoryBuffer.newIterator();
+ }
+ }
+
+ @VisibleForTesting
+ List<ChannelWithMeta> getSpillChannels() {
+ return spilledChannelIDs;
+ }
+
+ private class InMemoryBuffer {
+
+ private final AbstractRowDataSerializer<InternalRow> serializer;
+ private final ArrayList<MemorySegment> recordBufferSegments;
+ private final SimpleCollectingOutputView recordCollector;
+
+ private long currentDataBufferOffset;
+ private int numBytesInLastBuffer;
+ private int numRecords = 0;
+
+ private InMemoryBuffer(AbstractRowDataSerializer<InternalRow> serializer) {
+ // serializer has states, so we must duplicate
+ this.serializer = (AbstractRowDataSerializer<InternalRow>) serializer.duplicate();
+ this.recordBufferSegments = new ArrayList<>();
+ this.recordCollector =
+ new SimpleCollectingOutputView(this.recordBufferSegments, pool, segmentSize);
+ }
+
+ private void reset() {
+ this.currentDataBufferOffset = 0;
+ this.numRecords = 0;
+ returnToSegmentPool();
+ this.recordCollector.reset();
+ }
+
+ private void returnToSegmentPool() {
+ pool.returnAll(this.recordBufferSegments);
+ this.recordBufferSegments.clear();
+ }
+
+ public boolean write(InternalRow row) throws IOException {
+ try {
+ this.serializer.serializeToPages(row, this.recordCollector);
+ currentDataBufferOffset = this.recordCollector.getCurrentOffset();
+ numBytesInLastBuffer = this.recordCollector.getCurrentPositionInSegment();
+ numRecords++;
+ return true;
+ } catch (EOFException e) {
+ return false;
+ }
+ }
+
+ private ArrayList<MemorySegment> getRecordBufferSegments() {
+ return recordBufferSegments;
+ }
+
+ private long getCurrentDataBufferOffset() {
+ return currentDataBufferOffset;
+ }
+
+ private int getNumRecordBuffers() {
+ int result = (int) (currentDataBufferOffset / segmentSize);
+ long mod = currentDataBufferOffset % segmentSize;
+ if (mod != 0) {
+ result += 1;
+ }
+ return result;
+ }
+
+ private int getNumBytesInLastBuffer() {
+ return numBytesInLastBuffer;
+ }
+
+ private InMemoryBufferIterator newIterator() {
+ RandomAccessInputView recordBuffer =
+ new RandomAccessInputView(
+ this.recordBufferSegments, segmentSize, numBytesInLastBuffer);
+ return new InMemoryBufferIterator(recordBuffer, serializer);
+ }
+ }
+
+ private static class InMemoryBufferIterator
+ implements MutableObjectIterator<BinaryRow>, Closeable {
+
+ private final RandomAccessInputView recordBuffer;
+ private final AbstractRowDataSerializer<InternalRow> serializer;
+
+ private InMemoryBufferIterator(
+ RandomAccessInputView recordBuffer,
+ AbstractRowDataSerializer<InternalRow> serializer) {
+ this.recordBuffer = recordBuffer;
+ this.serializer = serializer;
+ }
+
+ @Override
+ public BinaryRow next(BinaryRow reuse) throws IOException {
+ try {
+ return (BinaryRow) serializer.mapFromPages(reuse, recordBuffer);
+ } catch (EOFException e) {
+ return null;
+ }
+ }
+
+ @Override
+ public BinaryRow next() throws IOException {
+ throw new RuntimeException("Not support!");
+ }
+
+ @Override
+ public void close() {}
+ }
+}
diff --git a/paimon-core/src/test/java/org/apache/paimon/disk/ExternalBufferTest.java b/paimon-core/src/test/java/org/apache/paimon/disk/ExternalBufferTest.java
new file mode 100644
index 000000000..5bd9ad511
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/disk/ExternalBufferTest.java
@@ -0,0 +1,209 @@
+/*
+ * 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.disk;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.BinaryRowWriter;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.serializer.BinaryRowSerializer;
+import org.apache.paimon.memory.HeapMemorySegmentPool;
+
+import org.apache.commons.math3.random.RandomDataGenerator;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+
+import static org.apache.paimon.memory.MemorySegmentPool.DEFAULT_PAGE_SIZE;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Test for {@link ExternalBuffer}. */
+public class ExternalBufferTest {
+
+ @TempDir Path tempDir;
+
+ private IOManager ioManager;
+ private Random random;
+ private BinaryRowSerializer serializer;
+
+ @BeforeEach
+ public void before() {
+ this.ioManager = IOManager.create(tempDir.toString());
+ this.random = new Random();
+ this.serializer = new BinaryRowSerializer(1);
+ }
+
+ private ExternalBuffer newBuffer() {
+ return new ExternalBuffer(
+ ioManager,
+ new HeapMemorySegmentPool(2 * DEFAULT_PAGE_SIZE, DEFAULT_PAGE_SIZE),
+ this.serializer);
+ }
+
+ @Test
+ public void testLess() throws Exception {
+ ExternalBuffer buffer = newBuffer();
+
+ int number = 100;
+ List<Long> expected = insertMulti(buffer, number);
+ assertThat(number).isEqualTo(buffer.size());
+ assertBuffer(expected, buffer);
+ assertThat(0).isEqualTo(buffer.getSpillChannels().size());
+
+ // repeat read
+ assertBuffer(expected, buffer);
+ buffer.newIterator();
+ assertBuffer(expected, buffer);
+ buffer.reset();
+ }
+
+ @Test
+ public void testSpill() throws Exception {
+ ExternalBuffer buffer = newBuffer();
+
+ int number = 5000; // 16 * 5000
+ List<Long> expected = insertMulti(buffer, number);
+ assertThat(number).isEqualTo(buffer.size());
+ assertBuffer(expected, buffer);
+ assertThat(buffer.getSpillChannels().size()).isGreaterThan(0);
+
+ // repeat read
+ assertBuffer(expected, buffer);
+ buffer.newIterator();
+ assertBuffer(expected, buffer);
+ buffer.reset();
+ }
+
+ @Test
+ public void testBufferReset() throws Exception {
+ ExternalBuffer buffer = newBuffer();
+
+ // less
+ insertMulti(buffer, 10);
+ buffer.reset();
+ assertThat(0).isEqualTo(buffer.size());
+
+ // not spill
+ List<Long> expected = insertMulti(buffer, 100);
+ assertThat(100).isEqualTo(buffer.size());
+ assertBuffer(expected, buffer);
+ buffer.reset();
+
+ // spill
+ expected = insertMulti(buffer, 2500);
+ assertThat(2500).isEqualTo(buffer.size());
+ assertBuffer(expected, buffer);
+ buffer.reset();
+ }
+
+ @Test
+ public void testBufferResetWithSpill() throws Exception {
+ int inMemoryThreshold = 20;
+ ExternalBuffer buffer = newBuffer();
+
+ // spill
+ List<Long> expected = insertMulti(buffer, 5000);
+ assertThat(5000).isEqualTo(buffer.size());
+ assertBuffer(expected, buffer);
+ buffer.reset();
+
+ // spill, but not read the values
+ insertMulti(buffer, 5000);
+ buffer.newIterator();
+ assertThat(5000).isEqualTo(buffer.size());
+ buffer.reset();
+
+ // not spill
+ expected = insertMulti(buffer, inMemoryThreshold / 2);
+ assertBuffer(expected, buffer);
+ buffer.reset();
+ assertThat(0).isEqualTo(buffer.size());
+
+ // less
+ expected = insertMulti(buffer, 100);
+ assertThat(100).isEqualTo(buffer.size());
+ assertBuffer(expected, buffer);
+ buffer.reset();
+ }
+
+ @Test
+ public void testHugeRecord() {
+ ExternalBuffer buffer =
+ new ExternalBuffer(
+ ioManager,
+ new HeapMemorySegmentPool(3 * DEFAULT_PAGE_SIZE, DEFAULT_PAGE_SIZE),
+ new BinaryRowSerializer(1));
+ assertThatThrownBy(() -> writeHuge(buffer)).isInstanceOf(IOException.class);
+ buffer.reset();
+ }
+
+ private void writeHuge(ExternalBuffer buffer) throws IOException {
+ BinaryRow row = new BinaryRow(1);
+ BinaryRowWriter writer = new BinaryRowWriter(row);
+ writer.reset();
+ RandomDataGenerator random = new RandomDataGenerator();
+ writer.writeString(0, BinaryString.fromString(random.nextHexString(500000)));
+ writer.complete();
+ buffer.add(row);
+ }
+
+ private void assertBuffer(List<Long> expected, ExternalBuffer buffer) {
+ ExternalBuffer.BufferIterator iterator = buffer.newIterator();
+ assertBuffer(expected, iterator);
+ iterator.close();
+ }
+
+ private void assertBuffer(List<Long> expected, ExternalBuffer.BufferIterator iterator) {
+ List<Long> values = new ArrayList<>();
+ while (iterator.advanceNext()) {
+ values.add(iterator.getRow().getLong(0));
+ }
+ assertThat(values).isEqualTo(expected);
+ }
+
+ private List<Long> insertMulti(ExternalBuffer buffer, int cnt) throws IOException {
+ ArrayList<Long> expected = new ArrayList<>(cnt);
+ insertMulti(buffer, cnt, expected);
+ buffer.complete();
+ return expected;
+ }
+
+ private void insertMulti(ExternalBuffer buffer, int cnt, List<Long> expected)
+ throws IOException {
+ for (int i = 0; i < cnt; i++) {
+ expected.add(randomInsert(buffer));
+ }
+ }
+
+ private long randomInsert(ExternalBuffer buffer) throws IOException {
+ long l = random.nextLong();
+ BinaryRow row = new BinaryRow(1);
+ BinaryRowWriter writer = new BinaryRowWriter(row);
+ writer.reset();
+ writer.writeLong(0, l);
+ writer.complete();
+ buffer.add(row);
+ return l;
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/cdc/CdcSinkBuilder.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/cdc/CdcSinkBuilder.java
index 4f1745406..1bda75695 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/cdc/CdcSinkBuilder.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/cdc/CdcSinkBuilder.java
@@ -21,7 +21,6 @@ package org.apache.paimon.flink.sink.cdc;
import org.apache.paimon.annotation.Experimental;
import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.Identifier;
-import org.apache.paimon.flink.sink.index.GlobalDynamicCdcBucketSink;
import org.apache.paimon.flink.utils.SingleOutputStreamOperatorUtils;
import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.table.BucketMode;
@@ -119,9 +118,6 @@ public class CdcSinkBuilder<T> {
return buildForFixedBucket(parsed);
case DYNAMIC:
return new CdcDynamicBucketSink((FileStoreTable) table).build(parsed, parallelism);
- case GLOBAL_DYNAMIC:
- return new GlobalDynamicCdcBucketSink((FileStoreTable) table)
- .build(parsed, parallelism);
case UNAWARE:
default:
throw new UnsupportedOperationException("Unsupported bucket mode: " + bucketMode);
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncDatabaseSinkBuilder.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncDatabaseSinkBuilder.java
index c5b716e45..a3dc7beeb 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncDatabaseSinkBuilder.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncDatabaseSinkBuilder.java
@@ -22,7 +22,6 @@ import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.flink.action.cdc.DatabaseSyncMode;
import org.apache.paimon.flink.sink.FlinkStreamPartitioner;
-import org.apache.paimon.flink.sink.index.GlobalDynamicCdcBucketSink;
import org.apache.paimon.flink.utils.SingleOutputStreamOperatorUtils;
import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.table.BucketMode;
@@ -196,8 +195,6 @@ public class FlinkCdcSyncDatabaseSinkBuilder<T> {
new CdcDynamicBucketSink(table).build(parsedForTable, parallelism);
break;
case GLOBAL_DYNAMIC:
- new GlobalDynamicCdcBucketSink(table).build(parsedForTable, parallelism);
- break;
case UNAWARE:
default:
throw new UnsupportedOperationException(
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/index/GlobalDynamicBucketSink.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/index/GlobalDynamicBucketSink.java
index 5f98f3b05..26cfc9b41 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/index/GlobalDynamicBucketSink.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/index/GlobalDynamicBucketSink.java
@@ -43,11 +43,10 @@ import javax.annotation.Nullable;
import java.util.List;
import java.util.Map;
import java.util.UUID;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
import static org.apache.paimon.flink.LogicalTypeConversion.toLogicalType;
import static org.apache.paimon.flink.sink.FlinkStreamPartitioner.partition;
+import static org.apache.paimon.flink.sink.index.IndexBootstrap.bootstrapType;
/** Sink for global dynamic bucket table. */
public class GlobalDynamicBucketSink extends FlinkWriteSink<Tuple2<RowData, Integer>> {
@@ -71,13 +70,10 @@ public class GlobalDynamicBucketSink extends FlinkWriteSink<Tuple2<RowData, Inte
TableSchema schema = table.schema();
RowType rowType = schema.logicalRowType();
List<String> primaryKeys = schema.primaryKeys();
- List<String> partitionKeys = schema.partitionKeys();
RowDataSerializer rowSerializer = new RowDataSerializer(toLogicalType(rowType));
- RowType keyPartType =
- schema.projectedLogicalRowType(
- Stream.concat(primaryKeys.stream(), partitionKeys.stream())
- .collect(Collectors.toList()));
- RowDataSerializer keyPartSerializer = new RowDataSerializer(toLogicalType(keyPartType));
+
+ RowType bootstrapType = bootstrapType(schema);
+ RowDataSerializer bootstrapSerializer = new RowDataSerializer(toLogicalType(bootstrapType));
// Topology:
// input -- bootstrap -- shuffle by key hash --> bucket-assigner -- shuffle by bucket -->
@@ -88,7 +84,7 @@ public class GlobalDynamicBucketSink extends FlinkWriteSink<Tuple2<RowData, Inte
"INDEX_BOOTSTRAP",
new InternalTypeInfo<>(
new KeyWithRowSerializer<>(
- keyPartSerializer, rowSerializer)),
+ bootstrapSerializer, rowSerializer)),
new IndexBootstrapOperator<>(
new IndexBootstrap(table), FlinkRowData::new))
.setParallelism(input.getParallelism());
@@ -100,7 +96,7 @@ public class GlobalDynamicBucketSink extends FlinkWriteSink<Tuple2<RowData, Inte
}
KeyPartRowChannelComputer channelComputer =
- new KeyPartRowChannelComputer(rowType, keyPartType, primaryKeys);
+ new KeyPartRowChannelComputer(rowType, bootstrapType, primaryKeys);
DataStream<Tuple2<KeyPartOrRow, RowData>> partitionByKeyHash =
partition(bootstraped, channelComputer, assignerParallelism);
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/index/GlobalDynamicCdcBucketSink.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/index/GlobalDynamicCdcBucketSink.java
deleted file mode 100644
index 06bb5a147..000000000
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/index/GlobalDynamicCdcBucketSink.java
+++ /dev/null
@@ -1,133 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.paimon.flink.sink.index;
-
-import org.apache.paimon.data.GenericRow;
-import org.apache.paimon.flink.sink.ChannelComputer;
-import org.apache.paimon.flink.sink.Committable;
-import org.apache.paimon.flink.sink.FlinkWriteSink;
-import org.apache.paimon.flink.sink.StoreSinkWrite;
-import org.apache.paimon.flink.sink.cdc.CdcDynamicBucketWriteOperator;
-import org.apache.paimon.flink.sink.cdc.CdcHashKeyChannelComputer;
-import org.apache.paimon.flink.sink.cdc.CdcRecord;
-import org.apache.paimon.flink.sink.cdc.CdcRecordUtils;
-import org.apache.paimon.flink.sink.cdc.CdcWithBucketChannelComputer;
-import org.apache.paimon.schema.TableSchema;
-import org.apache.paimon.table.FileStoreTable;
-import org.apache.paimon.types.RowType;
-import org.apache.paimon.utils.RowDataToObjectArrayConverter;
-
-import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
-import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.api.java.typeutils.EnumTypeInfo;
-import org.apache.flink.api.java.typeutils.TupleTypeInfo;
-import org.apache.flink.streaming.api.datastream.DataStream;
-import org.apache.flink.streaming.api.datastream.DataStreamSink;
-import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
-
-import javax.annotation.Nullable;
-
-import java.util.List;
-import java.util.UUID;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
-
-import static org.apache.paimon.flink.sink.FlinkStreamPartitioner.partition;
-
-/** Sink for global dynamic bucket table and {@link CdcRecord} inputs. */
-public class GlobalDynamicCdcBucketSink extends FlinkWriteSink<Tuple2<CdcRecord, Integer>> {
-
- private static final long serialVersionUID = 1L;
-
- public GlobalDynamicCdcBucketSink(FileStoreTable table) {
- super(table, null);
- }
-
- @Override
- protected OneInputStreamOperator<Tuple2<CdcRecord, Integer>, Committable> createWriteOperator(
- StoreSinkWrite.Provider writeProvider, String commitUser) {
- return new CdcDynamicBucketWriteOperator(table, writeProvider, commitUser);
- }
-
- public DataStreamSink<?> build(DataStream<CdcRecord> input, @Nullable Integer parallelism) {
- String initialCommitUser = UUID.randomUUID().toString();
-
- TableSchema schema = table.schema();
- List<String> primaryKeys = schema.primaryKeys();
- List<String> partitionKeys = schema.partitionKeys();
- RowType keyPartType =
- schema.projectedLogicalRowType(
- Stream.concat(primaryKeys.stream(), partitionKeys.stream())
- .collect(Collectors.toList()));
- List<String> keyPartNames = keyPartType.getFieldNames();
- RowDataToObjectArrayConverter keyPartConverter =
- new RowDataToObjectArrayConverter(keyPartType);
-
- // Topology:
- // input -- bootstrap -- shuffle by key hash --> bucket-assigner -- shuffle by bucket -->
- // writer --> committer
-
- TupleTypeInfo<Tuple2<KeyPartOrRow, CdcRecord>> tuple2TupleType =
- new TupleTypeInfo<>(new EnumTypeInfo<>(KeyPartOrRow.class), input.getType());
- DataStream<Tuple2<KeyPartOrRow, CdcRecord>> bootstraped =
- input.transform(
- "INDEX_BOOTSTRAP",
- tuple2TupleType,
- new IndexBootstrapOperator<>(
- new IndexBootstrap(table),
- row ->
- CdcRecordUtils.fromGenericRow(
- GenericRow.ofKind(
- row.getRowKind(),
- keyPartConverter.convert(row)),
- keyPartNames)))
- .setParallelism(input.getParallelism());
-
- // 1. shuffle by key hash
- Integer assignerParallelism = table.coreOptions().dynamicBucketAssignerParallelism();
- if (assignerParallelism == null) {
- assignerParallelism = parallelism;
- }
-
- ChannelComputer<Tuple2<KeyPartOrRow, CdcRecord>> channelComputer =
- ChannelComputer.transform(
- new CdcHashKeyChannelComputer(schema), tuple2 -> tuple2.f1);
- DataStream<Tuple2<KeyPartOrRow, CdcRecord>> partitionByKeyHash =
- partition(bootstraped, channelComputer, assignerParallelism);
-
- // 2. bucket-assigner
- TupleTypeInfo<Tuple2<CdcRecord, Integer>> rowWithBucketType =
- new TupleTypeInfo<>(input.getType(), BasicTypeInfo.INT_TYPE_INFO);
- DataStream<Tuple2<CdcRecord, Integer>> bucketAssigned =
- partitionByKeyHash
- .transform(
- "dynamic-bucket-assigner",
- rowWithBucketType,
- GlobalIndexAssignerOperator.forCdcRecord(table))
- .setParallelism(partitionByKeyHash.getParallelism());
-
- // 3. shuffle by bucket
-
- DataStream<Tuple2<CdcRecord, Integer>> partitionByBucket =
- partition(bucketAssigned, new CdcWithBucketChannelComputer(schema), parallelism);
-
- // 4. writer and committer
- return sinkFrom(partitionByBucket, initialCommitUser);
- }
-}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/index/GlobalIndexAssigner.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/index/GlobalIndexAssigner.java
index 2d3be0a12..78340b246 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/index/GlobalIndexAssigner.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/index/GlobalIndexAssigner.java
@@ -57,7 +57,8 @@ public class GlobalIndexAssigner<T> implements Serializable {
private final AbstractFileStoreTable table;
private final SerializableFunction<TableSchema, PartitionKeyExtractor<T>> extractorFunction;
private final SerializableFunction<TableSchema, PartitionKeyExtractor<T>>
- keyPartExtractorFunction;
+ bootstrapExtractorFunction;
+ private final SerializableFunction<T, Integer> bootstrapBucketFunction;
private final SerBiFunction<T, BinaryRow, T> setPartition;
private final SerBiFunction<T, RowKind, T> setRowKind;
@@ -78,12 +79,14 @@ public class GlobalIndexAssigner<T> implements Serializable {
public GlobalIndexAssigner(
Table table,
SerializableFunction<TableSchema, PartitionKeyExtractor<T>> extractorFunction,
- SerializableFunction<TableSchema, PartitionKeyExtractor<T>> keyPartExtractorFunction,
+ SerializableFunction<TableSchema, PartitionKeyExtractor<T>> bootstrapExtractorFunction,
+ SerializableFunction<T, Integer> bootstrapBucketFunction,
SerBiFunction<T, BinaryRow, T> setPartition,
SerBiFunction<T, RowKind, T> setRowKind) {
this.table = (AbstractFileStoreTable) table;
this.extractorFunction = extractorFunction;
- this.keyPartExtractorFunction = keyPartExtractorFunction;
+ this.bootstrapExtractorFunction = bootstrapExtractorFunction;
+ this.bootstrapBucketFunction = bootstrapBucketFunction;
this.setPartition = setPartition;
this.setRowKind = setRowKind;
}
@@ -97,7 +100,7 @@ public class GlobalIndexAssigner<T> implements Serializable {
CoreOptions coreOptions = table.coreOptions();
this.targetBucketRowNumber = (int) coreOptions.dynamicBucketTargetRowNum();
this.extractor = extractorFunction.apply(table.schema());
- this.keyPartExtractor = keyPartExtractorFunction.apply(table.schema());
+ this.keyPartExtractor = bootstrapExtractorFunction.apply(table.schema());
// state
Options options = coreOptions.toConfiguration();
@@ -164,9 +167,12 @@ public class GlobalIndexAssigner<T> implements Serializable {
public void bootstrap(T value) throws IOException {
BinaryRow partition = keyPartExtractor.partition(value);
- keyIndex.put(
- keyPartExtractor.trimmedPrimaryKey(value),
- new PositiveIntInt(partMapping.index(partition), assignBucket(partition)));
+ BinaryRow key = keyPartExtractor.trimmedPrimaryKey(value);
+ int partId = partMapping.index(partition);
+ int bucket = bootstrapBucketFunction.apply(value);
+ bucketAssigner.bootstrapBucket(partition, bucket);
+ PositiveIntInt partAndBucket = new PositiveIntInt(partId, bucket);
+ keyIndex.put(key, partAndBucket);
}
private void processNewRecord(BinaryRow partition, int partId, BinaryRow key, T value)
@@ -207,6 +213,15 @@ public class GlobalIndexAssigner<T> implements Serializable {
private final Map<BinaryRow, TreeMap<Integer, Integer>> stats = new HashMap<>();
+ public void bootstrapBucket(BinaryRow part, int bucket) {
+ TreeMap<Integer, Integer> bucketMap = bucketMap(part);
+ Integer count = bucketMap.get(bucket);
+ if (count == null) {
+ count = 0;
+ }
+ bucketMap.put(bucket, count + 1);
+ }
+
public int assignBucket(BinaryRow part, Filter<Integer> filter, int maxCount) {
TreeMap<Integer, Integer> bucketMap = bucketMap(part);
for (Map.Entry<Integer, Integer> entry : bucketMap.entrySet()) {
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/index/GlobalIndexAssignerOperator.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/index/GlobalIndexAssignerOperator.java
index 5d5acd203..0c10b01ff 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/index/GlobalIndexAssignerOperator.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/index/GlobalIndexAssignerOperator.java
@@ -18,56 +18,81 @@
package org.apache.paimon.flink.sink.index;
-import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.serializer.InternalRowSerializer;
+import org.apache.paimon.disk.ExternalBuffer;
+import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.flink.FlinkRowData;
+import org.apache.paimon.flink.FlinkRowWrapper;
import org.apache.paimon.flink.sink.RowDataPartitionKeyExtractor;
-import org.apache.paimon.flink.sink.cdc.CdcRecord;
-import org.apache.paimon.flink.sink.cdc.CdcRecordPartitionKeyExtractor;
-import org.apache.paimon.flink.sink.cdc.CdcRecordUtils;
import org.apache.paimon.flink.utils.ProjectToRowDataFunction;
-import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.memory.HeapMemorySegmentPool;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.table.AbstractFileStoreTable;
import org.apache.paimon.table.Table;
import org.apache.paimon.types.RowType;
-import org.apache.paimon.utils.RowDataToObjectArrayConverter;
+import org.apache.paimon.utils.SerializableFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.runtime.state.StateInitializationContext;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.table.data.RowData;
import java.io.File;
import java.io.IOException;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
import java.util.concurrent.ThreadLocalRandom;
import static org.apache.paimon.flink.FlinkRowData.toFlinkRowKind;
/** A {@link OneInputStreamOperator} for {@link GlobalIndexAssigner}. */
public class GlobalIndexAssignerOperator<T> extends AbstractStreamOperator<Tuple2<T, Integer>>
- implements OneInputStreamOperator<Tuple2<KeyPartOrRow, T>, Tuple2<T, Integer>> {
+ implements OneInputStreamOperator<Tuple2<KeyPartOrRow, T>, Tuple2<T, Integer>>,
+ BoundedOneInput {
private static final long serialVersionUID = 1L;
+ private final Table table;
private final GlobalIndexAssigner<T> assigner;
+ private final SerializableFunction<T, InternalRow> toRow;
+ private final SerializableFunction<InternalRow, T> fromRow;
- public GlobalIndexAssignerOperator(GlobalIndexAssigner<T> assigner) {
+ private transient ExternalBuffer bootstrapBuffer;
+
+ public GlobalIndexAssignerOperator(
+ Table table,
+ GlobalIndexAssigner<T> assigner,
+ SerializableFunction<T, InternalRow> toRow,
+ SerializableFunction<InternalRow, T> fromRow) {
+ this.table = table;
this.assigner = assigner;
+ this.toRow = toRow;
+ this.fromRow = fromRow;
}
@Override
public void initializeState(StateInitializationContext context) throws Exception {
super.initializeState(context);
- File[] tmpDirs =
- getContainingTask().getEnvironment().getIOManager().getSpillingDirectories();
+ org.apache.flink.runtime.io.disk.iomanager.IOManager ioManager =
+ getContainingTask().getEnvironment().getIOManager();
+ File[] tmpDirs = ioManager.getSpillingDirectories();
File tmpDir = tmpDirs[ThreadLocalRandom.current().nextInt(tmpDirs.length)];
assigner.open(
tmpDir,
getRuntimeContext().getNumberOfParallelSubtasks(),
getRuntimeContext().getIndexOfThisSubtask(),
this::collect);
+ Options options = Options.fromMap(table.options());
+ long bufferSize = options.get(CoreOptions.WRITE_BUFFER_SIZE).getBytes();
+ long pageSize = options.get(CoreOptions.PAGE_SIZE).getBytes();
+ bootstrapBuffer =
+ new ExternalBuffer(
+ IOManager.create(ioManager.getSpillingDirectoriesPaths()),
+ new HeapMemorySegmentPool(bufferSize, (int) pageSize),
+ new InternalRowSerializer(table.rowType()));
}
@Override
@@ -80,11 +105,38 @@ public class GlobalIndexAssignerOperator<T> extends AbstractStreamOperator<Tuple
assigner.bootstrap(value);
break;
case ROW:
- assigner.process(value);
+ if (bootstrapBuffer != null) {
+ bootstrapBuffer.add(toRow.apply(value));
+ } else {
+ assigner.process(value);
+ }
break;
}
}
+ @Override
+ public void prepareSnapshotPreBarrier(long checkpointId) throws Exception {
+ endBootstrap();
+ }
+
+ @Override
+ public void endInput() throws Exception {
+ endBootstrap();
+ }
+
+ private void endBootstrap() throws Exception {
+ if (bootstrapBuffer != null) {
+ bootstrapBuffer.complete();
+ try (ExternalBuffer.BufferIterator iterator = bootstrapBuffer.newIterator()) {
+ while (iterator.advanceNext()) {
+ assigner.process(fromRow.apply(iterator.getRow()));
+ }
+ }
+ bootstrapBuffer.reset();
+ bootstrapBuffer = null;
+ }
+ }
+
private void collect(T value, int bucket) {
output.collect(new StreamRecord<>(new Tuple2<>(value, bucket)));
}
@@ -95,39 +147,22 @@ public class GlobalIndexAssignerOperator<T> extends AbstractStreamOperator<Tuple
}
public static GlobalIndexAssignerOperator<RowData> forRowData(Table table) {
- return new GlobalIndexAssignerOperator<>(createRowDataAssigner(table));
+ return new GlobalIndexAssignerOperator<>(
+ table, createRowDataAssigner(table), FlinkRowWrapper::new, FlinkRowData::new);
}
public static GlobalIndexAssigner<RowData> createRowDataAssigner(Table t) {
+ RowType bootstrapType = IndexBootstrap.bootstrapType(((AbstractFileStoreTable) t).schema());
+ int bucketIndex = bootstrapType.getFieldCount() - 1;
return new GlobalIndexAssigner<>(
t,
RowDataPartitionKeyExtractor::new,
KeyPartPartitionKeyExtractor::new,
+ row -> row.getInt(bucketIndex),
new ProjectToRowDataFunction(t.rowType(), t.partitionKeys()),
(rowData, rowKind) -> {
rowData.setRowKind(toFlinkRowKind(rowKind));
return rowData;
});
}
-
- public static GlobalIndexAssignerOperator<CdcRecord> forCdcRecord(Table table) {
- RowType partitionType = ((FileStoreTable) table).schema().logicalPartitionType();
- List<String> partitionNames = partitionType.getFieldNames();
- RowDataToObjectArrayConverter converter = new RowDataToObjectArrayConverter(partitionType);
- GlobalIndexAssigner<CdcRecord> assigner =
- new GlobalIndexAssigner<>(
- table,
- CdcRecordPartitionKeyExtractor::new,
- CdcRecordPartitionKeyExtractor::new,
- (record, part) -> {
- CdcRecord partCdc =
- CdcRecordUtils.fromGenericRow(
- GenericRow.of(converter.convert(part)), partitionNames);
- Map<String, String> fields = new HashMap<>(record.fields());
- fields.putAll(partCdc.fields());
- return new CdcRecord(record.kind(), fields);
- },
- CdcRecord::setRowKind);
- return new GlobalIndexAssignerOperator<>(assigner);
- }
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/index/IndexBootstrap.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/index/IndexBootstrap.java
index 2df812007..95b4c1a94 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/index/IndexBootstrap.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/index/IndexBootstrap.java
@@ -18,26 +18,40 @@
package org.apache.paimon.flink.sink.index;
+import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.JoinedRow;
import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.Table;
import org.apache.paimon.table.source.AbstractInnerTableScan;
+import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.table.source.ReadBuilder;
+import org.apache.paimon.table.source.Split;
import org.apache.paimon.table.source.TableScan;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
import java.io.IOException;
import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
import java.util.function.Consumer;
import java.util.stream.Collectors;
import java.util.stream.Stream;
+import static org.apache.paimon.CoreOptions.SCAN_MODE;
+import static org.apache.paimon.CoreOptions.StartupMode.LATEST;
+
/** Bootstrap key index from Paimon table. */
public class IndexBootstrap implements Serializable {
private static final long serialVersionUID = 1L;
+ public static final String BUCKET_FIELD = "_BUCKET";
+
private final Table table;
public IndexBootstrap(Table table) {
@@ -56,14 +70,42 @@ public class IndexBootstrap implements Serializable {
.map(fieldNames::indexOf)
.mapToInt(Integer::intValue)
.toArray();
- ReadBuilder readBuilder = table.newReadBuilder().withProjection(projection);
+
+ // force using the latest scan mode
+ ReadBuilder readBuilder =
+ table.copy(Collections.singletonMap(SCAN_MODE.key(), LATEST.toString()))
+ .newReadBuilder()
+ .withProjection(projection);
AbstractInnerTableScan tableScan = (AbstractInnerTableScan) readBuilder.newScan();
TableScan.Plan plan =
tableScan.withBucketFilter(bucket -> bucket % numAssigners == assignId).plan();
- try (RecordReader<InternalRow> reader = readBuilder.newRead().createReader(plan)) {
- reader.forEachRemaining(collector);
+ for (Split split : plan.splits()) {
+ try (RecordReader<InternalRow> reader = readBuilder.newRead().createReader(split)) {
+ int bucket = ((DataSplit) split).bucket();
+ GenericRow bucketRow = GenericRow.of(bucket);
+ JoinedRow joinedRow = new JoinedRow();
+ reader.transform(row -> joinedRow.replace(row, bucketRow))
+ .forEachRemaining(collector);
+ }
}
}
+
+ public static RowType bootstrapType(TableSchema schema) {
+ List<String> primaryKeys = schema.primaryKeys();
+ List<String> partitionKeys = schema.partitionKeys();
+ List<DataField> bootstrapFields =
+ new ArrayList<>(
+ schema.projectedLogicalRowType(
+ Stream.concat(primaryKeys.stream(), partitionKeys.stream())
+ .collect(Collectors.toList()))
+ .getFields());
+ bootstrapFields.add(
+ new DataField(
+ RowType.currentHighestFieldId(bootstrapFields) + 1,
+ BUCKET_FIELD,
+ DataTypes.INT().notNull()));
+ return new RowType(bootstrapFields);
+ }
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/index/KeyPartRowChannelComputer.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/index/KeyPartRowChannelComputer.java
index 9535b3826..aff593007 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/index/KeyPartRowChannelComputer.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/index/KeyPartRowChannelComputer.java
@@ -45,7 +45,6 @@ public class KeyPartRowChannelComputer implements ChannelComputer<Tuple2<KeyPart
public KeyPartRowChannelComputer(
RowType rowType, RowType keyPartType, List<String> primaryKey) {
-
this.rowType = rowType;
this.keyPartType = keyPartType;
this.primaryKey = primaryKey;
@@ -65,4 +64,9 @@ public class KeyPartRowChannelComputer implements ChannelComputer<Tuple2<KeyPart
.apply(new FlinkRowWrapper(record.f1));
return Math.abs(key.hashCode() % numChannels);
}
+
+ @Override
+ public String toString() {
+ return "shuffle by key hash";
+ }
}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/GlobalDynamicBucketTableITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/GlobalDynamicBucketTableITCase.java
index 65c608ba4..794cb312f 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/GlobalDynamicBucketTableITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/GlobalDynamicBucketTableITCase.java
@@ -84,4 +84,23 @@ public class GlobalDynamicBucketTableITCase extends CatalogITCaseBase {
assertThat(sql("SELECT DISTINCT bucket FROM T$files"))
.containsExactlyInAnyOrder(Row.of(0), Row.of(1), Row.of(2));
}
+
+ @Test
+ public void testLargeRecords() {
+ sql(
+ "create table large_t (pt int, k int, v int, primary key (k) not enforced) partitioned by (pt) with ("
+ + "'bucket'='-1', "
+ + "'dynamic-bucket.target-row-num'='10000')");
+ sql(
+ "create temporary table src (pt int, k int, v int) with ("
+ + "'connector'='datagen', "
+ + "'number-of-rows'='100000', "
+ + "'fields.k.min'='0', "
+ + "'fields.k.max'='100000', "
+ + "'fields.pt.min'='0', "
+ + "'fields.pt.max'='1')");
+ sql("insert into large_t select * from src");
+ sql("insert into large_t select * from src");
+ assertThat(sql("select k, count(*) from large_t group by k having count(*) > 1")).isEmpty();
+ }
}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncTableSinkITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncTableSinkITCase.java
index 81c9d1495..346e48b37 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncTableSinkITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncTableSinkITCase.java
@@ -46,6 +46,7 @@ import org.apache.paimon.utils.TraceableFileIO;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
import org.junit.jupiter.api.io.TempDir;
@@ -76,6 +77,7 @@ public class FlinkCdcSyncTableSinkITCase extends AbstractTestBase {
innerTestRandomCdcEvents(-1, false);
}
+ @Disabled
@Test
@Timeout(120)
public void testRandomCdcEventsGlobalDynamicBucket() throws Exception {
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/index/IndexBootstrapTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/index/IndexBootstrapTest.java
index cd185d1de..aadec017e 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/index/IndexBootstrapTest.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/index/IndexBootstrapTest.java
@@ -65,15 +65,24 @@ public class IndexBootstrapTest extends TableTestBase {
GenericRow.of(7, 7));
IndexBootstrap indexBootstrap = new IndexBootstrap(table);
- List<Integer> result = new ArrayList<>();
- Consumer<InternalRow> consumer = row -> result.add(row.getInt(0));
+ List<GenericRow> result = new ArrayList<>();
+ Consumer<InternalRow> consumer =
+ row -> result.add(GenericRow.of(row.getInt(0), row.getInt(1)));
+
+ // output key and bucket
indexBootstrap.bootstrap(2, 0, consumer);
- assertThat(result).containsExactlyInAnyOrder(2, 3);
+ assertThat(result).containsExactlyInAnyOrder(GenericRow.of(2, 4), GenericRow.of(3, 0));
result.clear();
indexBootstrap.bootstrap(2, 1, consumer);
- assertThat(result).containsExactlyInAnyOrder(1, 4, 5, 6, 7);
+ assertThat(result)
+ .containsExactlyInAnyOrder(
+ GenericRow.of(1, 3),
+ GenericRow.of(4, 1),
+ GenericRow.of(5, 1),
+ GenericRow.of(6, 3),
+ GenericRow.of(7, 1));
result.clear();
}
}
[incubator-paimon] 10/11: [doc] Add presto sql 332 to download
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.5
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
commit d7b8107ed377a4275e3a992a69a9116ba4af62dd
Author: Jingsong <ji...@gmail.com>
AuthorDate: Wed Sep 6 11:02:15 2023 +0800
[doc] Add presto sql 332 to download
---
docs/content/engines/presto.md | 2 ++
1 file changed, 2 insertions(+)
diff --git a/docs/content/engines/presto.md b/docs/content/engines/presto.md
index 4687c963b..9a61285a5 100644
--- a/docs/content/engines/presto.md
+++ b/docs/content/engines/presto.md
@@ -43,6 +43,7 @@ Download the jar file with corresponding version.
| [0.236, 0.268) | [paimon-presto-0.236-{{< version >}}.jar](https://repo.maven.apache.org/maven2/org/apache/paimon/paimon-presto-0.236/{{< version >}}/paimon-presto-0.236-{{< version >}}.jar) |
| [0.268, 0.273) | [paimon-presto-0.268-{{< version >}}.jar](https://repo.maven.apache.org/maven2/org/apache/paimon/paimon-presto-0.268/{{< version >}}/paimon-presto-0.268-{{< version >}}.jar) |
| [0.273, latest] | [paimon-presto-0.273-{{< version >}}.jar](https://repo.maven.apache.org/maven2/org/apache/paimon/paimon-presto-0.273/{{< version >}}/paimon-presto-0.273-{{< version >}}.jar) |
+| Presto SQL 332 | [paimon-prestosql-332-{{< version >}}.jar](https://repo.maven.apache.org/maven2/org/apache/paimon/paimon-prestosql-332/{{< version >}}/paimon-prestosql-332-{{< version >}}.jar)|
{{< /stable >}}
@@ -53,6 +54,7 @@ Download the jar file with corresponding version.
| [0.236, 0.268) | [paimon-presto-0.236-{{< version >}}.jar](https://repository.apache.org/snapshots/org/apache/paimon/paimon-presto-0.236/{{< version >}}/) |
| [0.268, 0.273) | [paimon-presto-0.268-{{< version >}}.jar](https://repository.apache.org/snapshots/org/apache/paimon/paimon-presto-0.268/{{< version >}}/) |
| [0.273, latest] | [paimon-presto-0.273-{{< version >}}.jar](https://repository.apache.org/snapshots/org/apache/paimon/paimon-presto-0.273/{{< version >}}/) |
+| Presto SQL 332 | [paimon-prestosql-332-{{< version >}}.jar](https://repository.apache.org/snapshots/org/apache/paimon/paimon-prestosql-332/{{< version >}}/) |
{{< /unstable >}}
[incubator-paimon] 04/11: [doc] Fix minor SQL error in trino.md (#1882)
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.5
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
commit e6a98ba9524f5ab1150f1b4d4670a784dbc72678
Author: groobyming <65...@qq.com>
AuthorDate: Mon Aug 28 10:51:50 2023 +0800
[doc] Fix minor SQL error in trino.md (#1882)
---
docs/content/engines/trino.md | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/docs/content/engines/trino.md b/docs/content/engines/trino.md
index 9789b6201..a95e8169b 100644
--- a/docs/content/engines/trino.md
+++ b/docs/content/engines/trino.md
@@ -178,7 +178,7 @@ WITH (
changelog_producer = 'input'
)
-ALTER TABLE paimon.test_db.orders ADD COLUMN "shipping_address varchar;
+ALTER TABLE paimon.test_db.orders ADD COLUMN shipping_address varchar;
```
## Query
[incubator-paimon] 03/11: [doc] Update trino doc to trino 422 (#1884)
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.5
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
commit 23280843f1d3ef5e1d317f4902650934fca56f9d
Author: s7monk <34...@users.noreply.github.com>
AuthorDate: Mon Aug 28 10:50:40 2023 +0800
[doc] Update trino doc to trino 422 (#1884)
---
docs/content/engines/overview.md | 2 +-
docs/content/engines/trino.md | 34 ++++++++++++++++++----------------
docs/content/project/download.md | 2 ++
3 files changed, 21 insertions(+), 17 deletions(-)
diff --git a/docs/content/engines/overview.md b/docs/content/engines/overview.md
index 737d427cc..5f499fd65 100644
--- a/docs/content/engines/overview.md
+++ b/docs/content/engines/overview.md
@@ -38,7 +38,7 @@ Apache Spark and Apache Hive.
| Spark | 3.1 - 3.4 | ✅ | ✅ | ✅ | ✅ | ❌ | ❌ | ❌ |
| Hive | 2.1 - 3.1 | ✅ | ✅ | ✅ | ❌ | ❌ | ❌ | ❌ |
| Spark | 2.4 | ✅ | ❌ | ❌ | ❌ | ❌ | ❌ | ❌ |
-| Trino | 358 - 416 | ✅ | ❌ | ✅ | ✅ | ❌ | ❌ | ❌ |
+| Trino | 358 - 422 | ✅ | ❌ | ✅ | ✅ | ❌ | ❌ | ❌ |
| Presto | 0.236 - 0.280 | ✅ | ❌ | ✅ | ✅ | ❌ | ❌ | ❌ |
Ongoing engines:
diff --git a/docs/content/engines/trino.md b/docs/content/engines/trino.md
index 986973c35..9789b6201 100644
--- a/docs/content/engines/trino.md
+++ b/docs/content/engines/trino.md
@@ -38,27 +38,29 @@ Paimon currently supports Trino 358 and above.
Download the jar file with corresponding version.
-| Version | Jar |
-|---------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| [358, 368) | [paimon-trino-358-{{< version >}}.jar](https://repo.maven.apache.org/maven2/org/apache/paimon/paimon-trino-358/{{< version >}}/paimon-trino-358-{{< version >}}.jar) |
-| [368, 369) | [paimon-trino-368-{{< version >}}.jar](https://repo.maven.apache.org/maven2/org/apache/paimon/paimon-trino-368/{{< version >}}/paimon-trino-368-{{< version >}}.jar) |
-| [369, 370) | [paimon-trino-369-{{< version >}}.jar](https://repo.maven.apache.org/maven2/org/apache/paimon/paimon-trino-369/{{< version >}}/paimon-trino-369-{{< version >}}.jar) |
-| [370, 388) | [paimon-trino-370-{{< version >}}.jar](https://repo.maven.apache.org/maven2/org/apache/paimon/paimon-trino-370/{{< version >}}/paimon-trino-370-{{< version >}}.jar) |
-| [388, 393) | [paimon-trino-388-{{< version >}}.jar](https://repo.maven.apache.org/maven2/org/apache/paimon/paimon-trino-388/{{< version >}}/paimon-trino-388-{{< version >}}.jar) |
-| [393, latest] | [paimon-trino-393-{{< version >}}.jar](https://repo.maven.apache.org/maven2/org/apache/paimon/paimon-trino-393/{{< version >}}/paimon-trino-393-{{< version >}}.jar) |
+| Version | Jar |
+|---------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| [358, 368) | [paimon-trino-358-{{< version >}}.jar](https://repo.maven.apache.org/maven2/org/apache/paimon/paimon-trino-358/{{< version >}}/paimon-trino-358-{{< version >}}.jar) |
+| [368, 369) | [paimon-trino-368-{{< version >}}.jar](https://repo.maven.apache.org/maven2/org/apache/paimon/paimon-trino-368/{{< version >}}/paimon-trino-368-{{< version >}}.jar) |
+| [369, 370) | [paimon-trino-369-{{< version >}}.jar](https://repo.maven.apache.org/maven2/org/apache/paimon/paimon-trino-369/{{< version >}}/paimon-trino-369-{{< version >}}.jar) |
+| [370, 388) | [paimon-trino-370-{{< version >}}.jar](https://repo.maven.apache.org/maven2/org/apache/paimon/paimon-trino-370/{{< version >}}/paimon-trino-370-{{< version >}}.jar) |
+| [388, 393) | [paimon-trino-388-{{< version >}}.jar](https://repo.maven.apache.org/maven2/org/apache/paimon/paimon-trino-388/{{< version >}}/paimon-trino-388-{{< version >}}.jar) |
+| [393, 422] | [paimon-trino-393-{{< version >}}.jar](https://repo.maven.apache.org/maven2/org/apache/paimon/paimon-trino-393/{{< version >}}/paimon-trino-393-{{< version >}}.jar) |
+| [422, latest] | [paimon-trino-422-{{< version >}}.jar](https://repo.maven.apache.org/maven2/org/apache/paimon/paimon-trino-422/{{< version >}}/paimon-trino-422-{{< version >}}.jar) |
{{< /stable >}}
{{< unstable >}}
-| Version | Jar |
-|---------------|-------------------------------------------------------------------------------------------------------------------------------------|
-| [358, 368) | [paimon-trino-358-{{< version >}}.jar](https://repository.apache.org/snapshots/org/apache/paimon/paimon-trino-358/{{< version >}}/) |
-| [368, 369) | [paimon-trino-368-{{< version >}}.jar](https://repository.apache.org/snapshots/org/apache/paimon/paimon-trino-368/{{< version >}}/) |
-| [369, 370) | [paimon-trino-369-{{< version >}}.jar](https://repository.apache.org/snapshots/org/apache/paimon/paimon-trino-369/{{< version >}}/) |
-| [370, 388) | [paimon-trino-370-{{< version >}}.jar](https://repository.apache.org/snapshots/org/apache/paimon/paimon-trino-370/{{< version >}}/) |
-| [388, 393) | [paimon-trino-388-{{< version >}}.jar](https://repository.apache.org/snapshots/org/apache/paimon/paimon-trino-388/{{< version >}}/) |
-| [393, latest] | [paimon-trino-393-{{< version >}}.jar](https://repository.apache.org/snapshots/org/apache/paimon/paimon-trino-393/{{< version >}}/) |
+| Version | Jar |
+|------------|-------------------------------------------------------------------------------------------------------------------------------------|
+| [358, 368) | [paimon-trino-358-{{< version >}}.jar](https://repository.apache.org/snapshots/org/apache/paimon/paimon-trino-358/{{< version >}}/) |
+| [368, 369) | [paimon-trino-368-{{< version >}}.jar](https://repository.apache.org/snapshots/org/apache/paimon/paimon-trino-368/{{< version >}}/) |
+| [369, 370) | [paimon-trino-369-{{< version >}}.jar](https://repository.apache.org/snapshots/org/apache/paimon/paimon-trino-369/{{< version >}}/) |
+| [370, 388) | [paimon-trino-370-{{< version >}}.jar](https://repository.apache.org/snapshots/org/apache/paimon/paimon-trino-370/{{< version >}}/) |
+| [388, 393) | [paimon-trino-388-{{< version >}}.jar](https://repository.apache.org/snapshots/org/apache/paimon/paimon-trino-388/{{< version >}}/) |
+| [393, 422] | [paimon-trino-393-{{< version >}}.jar](https://repository.apache.org/snapshots/org/apache/paimon/paimon-trino-393/{{< version >}}/) |
+| [422, latest] | [paimon-trino-422-{{< version >}}.jar](https://repository.apache.org/snapshots/org/apache/paimon/paimon-trino-422/{{< version >}}/) |
{{< /unstable >}}
diff --git a/docs/content/project/download.md b/docs/content/project/download.md
index bbab0aeeb..771850866 100644
--- a/docs/content/project/download.md
+++ b/docs/content/project/download.md
@@ -59,6 +59,7 @@ This documentation is a guide for downloading Paimon Jars.
| Trino 370 | [paimon-trino-370-{{< version >}}.jar](https://repository.apache.org/snapshots/org/apache/paimon/paimon-trino-370/{{< version >}}/) |
| Trino 388 | [paimon-trino-388-{{< version >}}.jar](https://repository.apache.org/snapshots/org/apache/paimon/paimon-trino-388/{{< version >}}/) |
| Trino 393 | [paimon-trino-393-{{< version >}}.jar](https://repository.apache.org/snapshots/org/apache/paimon/paimon-trino-393/{{< version >}}/) |
+| Trino 422 | [paimon-trino-422-{{< version >}}.jar](https://repository.apache.org/snapshots/org/apache/paimon/paimon-trino-422/{{< version >}}/) |
{{< /unstable >}}
@@ -91,6 +92,7 @@ This documentation is a guide for downloading Paimon Jars.
| Trino 370 | [paimon-trino-370-{{< version >}}.jar](https://repo.maven.apache.org/maven2/org/apache/paimon/paimon-trino-370/{{< version >}}/paimon-trino-370-{{< version >}}.jar) |
| Trino 388 | [paimon-trino-388-{{< version >}}.jar](https://repo.maven.apache.org/maven2/org/apache/paimon/paimon-trino-388/{{< version >}}/paimon-trino-388-{{< version >}}.jar) |
| Trino 393 | [paimon-trino-393-{{< version >}}.jar](https://repo.maven.apache.org/maven2/org/apache/paimon/paimon-trino-393/{{< version >}}/paimon-trino-393-{{< version >}}.jar) |
+| Trino 422 | [paimon-trino-422-{{< version >}}.jar](https://repo.maven.apache.org/maven2/org/apache/paimon/paimon-trino-422/{{< version >}}/paimon-trino-422-{{< version >}}.jar) |
{{< /stable >}}
[incubator-paimon] 05/11: [doc] Document Spark truncate table
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.5
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
commit 461096ef600cb24e39a27486aa96206c4874a09f
Author: Jingsong <ji...@gmail.com>
AuthorDate: Mon Aug 28 12:12:05 2023 +0800
[doc] Document Spark truncate table
---
docs/content/how-to/writing-tables.md | 17 +++++++++++++----
1 file changed, 13 insertions(+), 4 deletions(-)
diff --git a/docs/content/how-to/writing-tables.md b/docs/content/how-to/writing-tables.md
index 12268e156..07c964261 100644
--- a/docs/content/how-to/writing-tables.md
+++ b/docs/content/how-to/writing-tables.md
@@ -224,20 +224,29 @@ INSERT OVERWRITE MyTable SELECT ...
{{< /tabs >}}
-## Purging tables
+## Truncate tables
-You can use `INSERT OVERWRITE` to purge tables by inserting empty value.
-
-{{< tabs "purge-tables-syntax" >}}
+{{< tabs "truncate-tables-syntax" >}}
{{< tab "Flink" >}}
+You can use `INSERT OVERWRITE` to purge tables by inserting empty value.
+
```sql
INSERT OVERWRITE MyTable /*+ OPTIONS('dynamic-partition-overwrite'='false') */ SELECT * FROM MyTable WHERE false;
```
{{< /tab >}}
+{{< tab "Spark" >}}
+
+```sql
+TRUNCATE TABLE MyTable;
+```
+
+{{< /tab >}}
+
+
{{< /tabs >}}
## Purging Partitions
[incubator-paimon] 07/11: [doc] Document warehouse dir for FlinkGenericCatalog
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.5
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
commit 10d933cfb5db6b3dda1fb4b505c3fa22e36e6c0c
Author: Jingsong <ji...@gmail.com>
AuthorDate: Thu Aug 31 18:17:07 2023 +0800
[doc] Document warehouse dir for FlinkGenericCatalog
---
docs/content/engines/flink.md | 5 +++++
1 file changed, 5 insertions(+)
diff --git a/docs/content/engines/flink.md b/docs/content/engines/flink.md
index 504872b87..05c13ae42 100644
--- a/docs/content/engines/flink.md
+++ b/docs/content/engines/flink.md
@@ -155,6 +155,11 @@ Flink Generic Tables (Kafka and other tables)!
In this mode, you should use 'connector' option for creating tables.
+{{< hint info >}}
+Paimon will use `hive.metastore.warehouse.dir` in your `hive-site.xml`, please use path with scheme.
+For example, `hdfs://...`. Otherwise, Paimon will use the local path.
+{{< /hint >}}
+
```sql
CREATE CATALOG my_catalog WITH (
'type'='paimon-generic',