Skip to content

Commit ae7055f

Browse files
AitoziStefanietry
Aitozi
authored and
Stefanietry
committed
[core] Support compact for chain table
1 parent 021cd92 commit ae7055f

5 files changed

Lines changed: 377 additions & 1 deletion

File tree

‎docs/content/primary-key-table/chain-table.md‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -87,7 +87,6 @@ Notice that:
8787
- Chain table is only supported for primary key table, which means you should define `bucket` and `bucket-key` for the table.
8888
- Chain table should ensure that the schema of each branch is consistent.
8989
- Only spark support now, flink will be supported later.
90-
- Chain compact is not supported for now, and it will be supported later.
9190
- Deletion vector is not supported for chain table.
9291

9392
After creating a chain table, you can read and write data in the following ways.

‎paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkProcedures.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020

2121
import org.apache.paimon.spark.procedure.AlterFunctionProcedure;
2222
import org.apache.paimon.spark.procedure.AlterViewDialectProcedure;
23+
import org.apache.paimon.spark.procedure.ChainMergeProcedure;
2324
import org.apache.paimon.spark.procedure.ClearConsumersProcedure;
2425
import org.apache.paimon.spark.procedure.CompactDatabaseProcedure;
2526
import org.apache.paimon.spark.procedure.CompactManifestProcedure;
@@ -123,6 +124,7 @@ private static Map<String, Supplier<ProcedureBuilder>> initProcedureBuilders() {
123124
"trigger_tag_automatic_creation", TriggerTagAutomaticCreationProcedure::builder);
124125
procedureBuilders.put("rewrite_file_index", RewriteFileIndexProcedure::builder);
125126
procedureBuilders.put("copy", CopyFilesProcedure::builder);
127+
procedureBuilders.put("chain_merge", ChainMergeProcedure::builder);
126128
return procedureBuilders.build();
127129
}
128130
}

‎paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/ChainMergeProcedure.java‎

Lines changed: 302 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,302 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing, software
13+
* distributed under the License is distributed on an "AS IS" BASIS,
14+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
* See the License for the specific language governing permissions and
16+
* limitations under the License.
17+
*/
18+
19+
package org.apache.paimon.spark.procedure;
20+
21+
import org.apache.paimon.disk.IOManager;
22+
import org.apache.paimon.predicate.Predicate;
23+
import org.apache.paimon.reader.RecordReader;
24+
import org.apache.paimon.reader.RecordReaderIterator;
25+
import org.apache.paimon.spark.SparkTable;
26+
import org.apache.paimon.spark.SparkUtils;
27+
import org.apache.paimon.spark.catalyst.analysis.expressions.ExpressionUtils;
28+
import org.apache.paimon.table.ChainGroupReadTable;
29+
import org.apache.paimon.table.FallbackReadFileStoreTable;
30+
import org.apache.paimon.table.FileStoreTable;
31+
import org.apache.paimon.table.Table;
32+
import org.apache.paimon.table.sink.BatchTableCommit;
33+
import org.apache.paimon.table.sink.BatchTableWrite;
34+
import org.apache.paimon.table.sink.BatchWriteBuilder;
35+
import org.apache.paimon.table.sink.CommitMessage;
36+
import org.apache.paimon.table.sink.CommitMessageSerializer;
37+
import org.apache.paimon.table.source.ChainSplit;
38+
import org.apache.paimon.table.source.Split;
39+
import org.apache.paimon.table.source.TableRead;
40+
import org.apache.paimon.utils.InternalRowPartitionComputer;
41+
import org.apache.paimon.utils.ParameterUtils;
42+
import org.apache.paimon.utils.Preconditions;
43+
import org.apache.paimon.utils.StringUtils;
44+
45+
import org.apache.spark.api.java.JavaRDD;
46+
import org.apache.spark.api.java.JavaSparkContext;
47+
import org.apache.spark.api.java.function.FlatMapFunction;
48+
import org.apache.spark.sql.catalyst.InternalRow;
49+
import org.apache.spark.sql.catalyst.expressions.Expression;
50+
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan;
51+
import org.apache.spark.sql.connector.catalog.Identifier;
52+
import org.apache.spark.sql.connector.catalog.TableCatalog;
53+
import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation;
54+
import org.apache.spark.sql.types.DataTypes;
55+
import org.apache.spark.sql.types.Metadata;
56+
import org.apache.spark.sql.types.StructField;
57+
import org.apache.spark.sql.types.StructType;
58+
import org.slf4j.Logger;
59+
import org.slf4j.LoggerFactory;
60+
61+
import java.util.ArrayList;
62+
import java.util.Iterator;
63+
import java.util.List;
64+
import java.util.Map;
65+
66+
import scala.Option;
67+
68+
import static org.apache.paimon.spark.utils.SparkProcedureUtils.toWhere;
69+
import static org.apache.paimon.utils.Preconditions.checkArgument;
70+
import static org.apache.spark.sql.types.DataTypes.StringType;
71+
72+
/**
73+
* Chain merge procedure. Usage:
74+
*
75+
* <pre><code>
76+
* CALL sys.compact(table => 'tableId', partitions => 'p1=0,p2=0', target_branch => 'snapshot')
77+
* </code></pre>
78+
*/
79+
public class ChainMergeProcedure extends BaseProcedure {
80+
81+
private static final Logger LOG = LoggerFactory.getLogger(ChainMergeProcedure.class);
82+
83+
private static final ProcedureParameter[] PARAMETERS =
84+
new ProcedureParameter[] {
85+
ProcedureParameter.required("table", StringType),
86+
ProcedureParameter.required("partitions", StringType),
87+
ProcedureParameter.optional("target_branch", StringType),
88+
};
89+
90+
private static final StructType OUTPUT_TYPE =
91+
new StructType(
92+
new StructField[] {
93+
new StructField("result", DataTypes.BooleanType, true, Metadata.empty())
94+
});
95+
96+
protected ChainMergeProcedure(TableCatalog tableCatalog) {
97+
super(tableCatalog);
98+
}
99+
100+
@Override
101+
public ProcedureParameter[] parameters() {
102+
return PARAMETERS;
103+
}
104+
105+
@Override
106+
public StructType outputType() {
107+
return OUTPUT_TYPE;
108+
}
109+
110+
private boolean blank(InternalRow args, int index) {
111+
return args.isNullAt(index) || StringUtils.isNullOrWhitespaceOnly(args.getString(index));
112+
}
113+
114+
@Override
115+
public InternalRow[] call(InternalRow args) {
116+
Identifier tableIdent = toIdentifier(args.getString(0), PARAMETERS[0].name());
117+
String partitions = args.getString(1);
118+
SparkTable sparkTable = loadSparkTable(tableIdent);
119+
String targetBranch =
120+
blank(args, 2)
121+
? sparkTable.coreOptions().scanFallbackSnapshotBranch()
122+
: args.getString(2);
123+
List<Map<String, String>> compactPartitions =
124+
ParameterUtils.getPartitions(partitions.split(";"));
125+
validataChainMerge(sparkTable, targetBranch, partitions, compactPartitions);
126+
DataSourceV2Relation relation = createRelation(tableIdent);
127+
Expression condition =
128+
getPartitionCondition(relation, sparkTable.table(), toWhere(partitions));
129+
boolean executed =
130+
executeChainMerge(
131+
(FallbackReadFileStoreTable) sparkTable.getTable(),
132+
condition,
133+
relation,
134+
targetBranch);
135+
return new InternalRow[] {newInternalRow(executed)};
136+
}
137+
138+
public boolean executeChainMerge(
139+
FallbackReadFileStoreTable chainTable,
140+
Expression partCondition,
141+
DataSourceV2Relation relation,
142+
String targetBranch) {
143+
144+
// build scan for the specific partition
145+
Preconditions.checkArgument(
146+
chainTable.other() instanceof ChainGroupReadTable,
147+
"The chain merge should perform on the ChainFileStoreTable");
148+
149+
Option<Predicate> filter =
150+
ExpressionUtils.convertConditionToPaimonPredicate(
151+
partCondition,
152+
((LogicalPlan) relation).output(),
153+
chainTable.rowType(),
154+
false);
155+
156+
ChainGroupReadTable chainGroupReadTable = (ChainGroupReadTable) chainTable.other();
157+
ChainGroupReadTable.ChainTableBatchScan scan =
158+
(ChainGroupReadTable.ChainTableBatchScan) chainGroupReadTable.newScan();
159+
if (filter.isDefined()) {
160+
scan.withFilter(filter.get());
161+
}
162+
List<Split> splits = scan.plan().splits();
163+
164+
if (splits.isEmpty()) {
165+
LOG.info("The target partition={} is empty", partCondition);
166+
return false;
167+
}
168+
Preconditions.checkArgument(
169+
splits.stream().allMatch(s -> (s instanceof ChainSplit)),
170+
"The chain merge only accepts ChainDataSplit");
171+
172+
// build snapshot branch write builder with static partition overwrite
173+
FileStoreTable targetTable = ((ChainGroupReadTable) chainTable.other()).wrapped();
174+
checkArgument(
175+
targetBranch.equals(targetTable.coreOptions().branch()),
176+
"chain_merge should merge to snapshot branch");
177+
InternalRowPartitionComputer computer =
178+
new InternalRowPartitionComputer(
179+
chainTable.coreOptions().partitionDefaultName(),
180+
chainTable.schema().logicalPartitionType(),
181+
chainTable.schema().partitionKeys().toArray(new String[0]),
182+
chainTable.coreOptions().legacyPartitionName());
183+
Map<String, String> targetPartition =
184+
computer.generatePartValues(((ChainSplit) splits.get(0)).logicalPartition());
185+
186+
LOG.info(
187+
"Direct chain_merge plan built, splits: {}, target partition: {}, target branch: {}",
188+
splits.size(),
189+
targetPartition,
190+
targetBranch);
191+
BatchWriteBuilder writeBuilder =
192+
targetTable.newBatchWriteBuilder().withOverwrite(targetPartition);
193+
JavaSparkContext javaSparkContext = new JavaSparkContext(spark().sparkContext());
194+
195+
JavaRDD<byte[]> commitMessageJavaRDD =
196+
javaSparkContext
197+
.parallelize(splits)
198+
.mapPartitions(
199+
(FlatMapFunction<Iterator<Split>, byte[]>)
200+
splitIterator -> {
201+
List<byte[]> serializedMessages = new ArrayList<>();
202+
IOManager ioManager = SparkUtils.createIOManager();
203+
BatchTableWrite write = writeBuilder.newWrite();
204+
write.withIOManager(ioManager);
205+
while (splitIterator.hasNext()) {
206+
Split split = splitIterator.next();
207+
try {
208+
TableRead read =
209+
chainGroupReadTable
210+
.newRead()
211+
.withIOManager(ioManager);
212+
RecordReader<org.apache.paimon.data.InternalRow>
213+
reader = read.createReader(split);
214+
try (RecordReader<
215+
org.apache.paimon.data
216+
.InternalRow>
217+
rr = reader) {
218+
RecordReaderIterator<
219+
org.apache.paimon.data
220+
.InternalRow>
221+
it = new RecordReaderIterator<>(rr);
222+
org.apache.paimon.data.InternalRow row;
223+
while ((row = it.next()) != null) {
224+
write.write(row);
225+
}
226+
}
227+
CommitMessageSerializer serializer =
228+
new CommitMessageSerializer();
229+
List<CommitMessage> messages =
230+
write.prepareCommit();
231+
for (CommitMessage commitMessage : messages) {
232+
serializedMessages.add(
233+
serializer.serialize(
234+
commitMessage));
235+
}
236+
} finally {
237+
write.close();
238+
ioManager.close();
239+
}
240+
}
241+
return serializedMessages.iterator();
242+
});
243+
244+
try (BatchTableCommit commit = writeBuilder.newCommit()) {
245+
CommitMessageSerializer serializer = new CommitMessageSerializer();
246+
List<byte[]> serializedMessages = commitMessageJavaRDD.collect();
247+
List<CommitMessage> messages = new ArrayList<>(serializedMessages.size());
248+
for (byte[] serializedMessage : serializedMessages) {
249+
messages.add(serializer.deserialize(serializer.getVersion(), serializedMessage));
250+
}
251+
commit.commit(messages);
252+
} catch (Exception e) {
253+
throw new RuntimeException(e);
254+
}
255+
return true;
256+
}
257+
258+
private Expression getPartitionCondition(
259+
DataSourceV2Relation relation, Table table, String where) {
260+
Expression condition = null;
261+
if (!StringUtils.isNullOrWhitespaceOnly(where)) {
262+
condition = ExpressionUtils.resolveFilter(spark(), relation, where);
263+
checkArgument(
264+
ExpressionUtils.isValidPredicate(
265+
spark(), condition, table.partitionKeys().toArray(new String[0])),
266+
"Only partition predicate is supported, your predicate is %s, but partition keys are %s",
267+
condition,
268+
table.partitionKeys());
269+
}
270+
return condition;
271+
}
272+
273+
public static ProcedureBuilder builder() {
274+
return new BaseProcedure.Builder<ChainMergeProcedure>() {
275+
@Override
276+
public ChainMergeProcedure doBuild() {
277+
return new ChainMergeProcedure(tableCatalog());
278+
}
279+
};
280+
}
281+
282+
private void validataChainMerge(
283+
SparkTable sparkTable,
284+
String targetBranch,
285+
String partitions,
286+
List<Map<String, String>> compactPartitions) {
287+
checkArgument(
288+
sparkTable.coreOptions().isChainTable(), "chain_merge only supports chain table");
289+
checkArgument(
290+
targetBranch.equals(sparkTable.coreOptions().scanFallbackSnapshotBranch()),
291+
"chain_merge should merge to snapshot branch");
292+
checkArgument(
293+
sparkTable.getTable() instanceof FallbackReadFileStoreTable,
294+
"The chain merge should perform on the chain table");
295+
checkArgument(
296+
compactPartitions.size() == 1
297+
&& compactPartitions.get(0).size()
298+
== sparkTable.table().partitionKeys().size(),
299+
"chain_merge only supports one partition %s",
300+
partitions);
301+
}
302+
}

‎paimon-spark/paimon-spark-ut/pom.xml‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -95,6 +95,12 @@ under the License.
9595
<version>${project.version}</version>
9696
<scope>test</scope>
9797
</dependency>
98+
<dependency>
99+
<groupId>org.apache.paimon</groupId>
100+
<artifactId>paimon-spark-common_2.12</artifactId>
101+
<version>1.5-SNAPSHOT</version>
102+
<scope>test</scope>
103+
</dependency>
98104
</dependencies>
99105

100106
<build>

0 commit comments

Comments
 (0)