This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
The following commit(s) were added to refs/heads/master by this push:
new deec72ad7 [core] Support skip rewrite when upgrade in
ChangelogMergeTreeRewriter (#2751)
deec72ad7 is described below
commit deec72ad7a22447b03a30d503926835a15040db6
Author: Zouxxyy <[email protected]>
AuthorDate: Wed Jan 24 14:12:35 2024 +0800
[core] Support skip rewrite when upgrade in ChangelogMergeTreeRewriter
(#2751)
---
.../src/main/java/org/apache/paimon/KeyValue.java | 4 ++
.../compact/ChangelogMergeTreeRewriter.java | 56 ++++++++++++---
.../paimon/mergetree/compact/CompactRewriter.java | 23 +++++-
.../compact/FirstRowMergeTreeCompactRewriter.java | 5 ++
.../compact/FullChangelogMergeFunctionWrapper.java | 10 +--
.../FullChangelogMergeTreeCompactRewriter.java | 7 +-
.../LookupChangelogMergeFunctionWrapper.java | 11 +--
.../compact/LookupMergeTreeCompactRewriter.java | 5 ++
.../paimon/operation/KeyValueFileStoreWrite.java | 11 ++-
.../paimon/spark/sql/LookupCompactionTest.scala | 84 ++++++++++++++++++++++
10 files changed, 185 insertions(+), 31 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/KeyValue.java
b/paimon-core/src/main/java/org/apache/paimon/KeyValue.java
index 27b5cda7c..bbab367e4 100644
--- a/paimon-core/src/main/java/org/apache/paimon/KeyValue.java
+++ b/paimon-core/src/main/java/org/apache/paimon/KeyValue.java
@@ -95,6 +95,10 @@ public class KeyValue {
return valueKind;
}
+ public boolean isAdd() {
+ return valueKind.isAdd();
+ }
+
public InternalRow value() {
return value;
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/ChangelogMergeTreeRewriter.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/ChangelogMergeTreeRewriter.java
index efdca7944..4626d8fe0 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/ChangelogMergeTreeRewriter.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/ChangelogMergeTreeRewriter.java
@@ -18,6 +18,7 @@
package org.apache.paimon.mergetree.compact;
+import org.apache.paimon.CoreOptions;
import org.apache.paimon.KeyValue;
import org.apache.paimon.codegen.RecordEqualiser;
import org.apache.paimon.compact.CompactResult;
@@ -35,14 +36,21 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
+import java.util.stream.Collectors;
-/** A {@link MergeTreeCompactRewriter} which produces changelog files for the
compaction. */
+/**
+ * A {@link MergeTreeCompactRewriter} which produces changelog files while
performing compaction.
+ */
public abstract class ChangelogMergeTreeRewriter extends
MergeTreeCompactRewriter {
+ protected final int maxLevel;
+ protected final CoreOptions.MergeEngine mergeEngine;
protected final RecordEqualiser valueEqualiser;
protected final boolean changelogRowDeduplicate;
public ChangelogMergeTreeRewriter(
+ int maxLevel,
+ CoreOptions.MergeEngine mergeEngine,
KeyValueFileReaderFactory readerFactory,
KeyValueFileWriterFactory writerFactory,
Comparator<InternalRow> keyComparator,
@@ -51,6 +59,8 @@ public abstract class ChangelogMergeTreeRewriter extends
MergeTreeCompactRewrite
RecordEqualiser valueEqualiser,
boolean changelogRowDeduplicate) {
super(readerFactory, writerFactory, keyComparator, mfFactory,
mergeSorter);
+ this.maxLevel = maxLevel;
+ this.mergeEngine = mergeEngine;
this.valueEqualiser = valueEqualiser;
this.changelogRowDeduplicate = changelogRowDeduplicate;
}
@@ -83,14 +93,20 @@ public abstract class ChangelogMergeTreeRewriter extends
MergeTreeCompactRewrite
public CompactResult rewrite(
int outputLevel, boolean dropDelete, List<List<SortedRun>>
sections) throws Exception {
if (rewriteChangelog(outputLevel, dropDelete, sections)) {
- return rewriteChangelogCompaction(outputLevel, sections);
+ return rewriteChangelogCompaction(outputLevel, sections, true);
} else {
return rewriteCompaction(outputLevel, dropDelete, sections);
}
}
+ /**
+ * Rewrite and produce changelog at the same time.
+ *
+ * @param rewriteCompactFile whether to rewrite compact file
+ */
private CompactResult rewriteChangelogCompaction(
- int outputLevel, List<List<SortedRun>> sections) throws Exception {
+ int outputLevel, List<List<SortedRun>> sections, boolean
rewriteCompactFile)
+ throws Exception {
List<ConcatRecordReader.ReaderSupplier<ChangelogResult>>
sectionReaders = new ArrayList<>();
for (List<SortedRun> section : sections) {
sectionReaders.add(
@@ -109,12 +125,14 @@ public abstract class ChangelogMergeTreeRewriter extends
MergeTreeCompactRewrite
try {
iterator = new
RecordReaderIterator<>(ConcatRecordReader.create(sectionReaders));
- compactFileWriter =
writerFactory.createRollingMergeTreeFileWriter(outputLevel);
+ if (rewriteCompactFile) {
+ compactFileWriter =
writerFactory.createRollingMergeTreeFileWriter(outputLevel);
+ }
changelogFileWriter =
writerFactory.createRollingChangelogFileWriter(outputLevel);
while (iterator.hasNext()) {
ChangelogResult result = iterator.next();
- if (result.result() != null) {
+ if (rewriteCompactFile && result.result() != null) {
compactFileWriter.write(result.result());
}
for (KeyValue kv : result.changelogs()) {
@@ -133,10 +151,15 @@ public abstract class ChangelogMergeTreeRewriter extends
MergeTreeCompactRewrite
}
}
- return new CompactResult(
- extractFilesFromSections(sections),
- compactFileWriter.result(),
- changelogFileWriter.result());
+ List<DataFileMeta> before = extractFilesFromSections(sections);
+ List<DataFileMeta> after =
+ rewriteCompactFile
+ ? compactFileWriter.result()
+ : before.stream()
+ .map(x -> x.upgrade(outputLevel))
+ .collect(Collectors.toList());
+
+ return new CompactResult(before, after, changelogFileWriter.result());
}
@Override
@@ -145,9 +168,22 @@ public abstract class ChangelogMergeTreeRewriter extends
MergeTreeCompactRewrite
return rewriteChangelogCompaction(
outputLevel,
Collections.singletonList(
-
Collections.singletonList(SortedRun.fromSingle(file))));
+
Collections.singletonList(SortedRun.fromSingle(file))),
+ !rewriteCompactFileForUpgrade(outputLevel));
} else {
return super.upgrade(outputLevel, file);
}
}
+
+ /**
+ * For upgrade, there are two scenarios can skip rewrite compact file.
+ *
+ * <ul>
+ * <li>mergeEngine == DEDUPLICATE, no need to consider previous records
+ * <li>outputLevel == maxLevel, no previous records
+ * </ul>
+ */
+ private boolean rewriteCompactFileForUpgrade(int outputLevel) {
+ return mergeEngine == CoreOptions.MergeEngine.DEDUPLICATE ||
outputLevel == maxLevel;
+ }
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/CompactRewriter.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/CompactRewriter.java
index 03ff9ed44..af58720d9 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/CompactRewriter.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/CompactRewriter.java
@@ -25,11 +25,32 @@ import org.apache.paimon.mergetree.SortedRun;
import java.io.Closeable;
import java.util.List;
-/** Rewrite sections to the files. */
+/** Rewrite sections to new level. */
public interface CompactRewriter extends Closeable {
+ /**
+ * Rewrite sections to new level.
+ *
+ * @param outputLevel new level
+ * @param dropDelete whether to drop the deletion, see {@link
+ * MergeTreeCompactManager#triggerCompaction}
+ * @param sections list of sections (section is a list of {@link
SortedRun}s, and key intervals
+ * between sections do not overlap)
+ * @return compaction result
+ * @throws Exception exception
+ */
CompactResult rewrite(int outputLevel, boolean dropDelete,
List<List<SortedRun>> sections)
throws Exception;
+ /**
+ * Upgrade file to new level, usually file data is not rewritten, only the
metadata is updated.
+ * But in some certain scenarios, we must rewrite file too, e.g. {@link
+ * ChangelogMergeTreeRewriter}
+ *
+ * @param outputLevel new level
+ * @param file file to be updated
+ * @return compaction result
+ * @throws Exception exception
+ */
CompactResult upgrade(int outputLevel, DataFileMeta file) throws Exception;
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/FirstRowMergeTreeCompactRewriter.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/FirstRowMergeTreeCompactRewriter.java
index 9ae4da9b9..7336ade38 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/FirstRowMergeTreeCompactRewriter.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/FirstRowMergeTreeCompactRewriter.java
@@ -18,6 +18,7 @@
package org.apache.paimon.mergetree.compact;
+import org.apache.paimon.CoreOptions;
import org.apache.paimon.KeyValue;
import org.apache.paimon.annotation.VisibleForTesting;
import org.apache.paimon.codegen.RecordEqualiser;
@@ -46,6 +47,8 @@ public class FirstRowMergeTreeCompactRewriter extends
ChangelogMergeTreeRewriter
private final ContainsLevels containsLevels;
public FirstRowMergeTreeCompactRewriter(
+ int maxLevel,
+ CoreOptions.MergeEngine mergeEngine,
ContainsLevels containsLevels,
KeyValueFileReaderFactory readerFactory,
KeyValueFileWriterFactory writerFactory,
@@ -55,6 +58,8 @@ public class FirstRowMergeTreeCompactRewriter extends
ChangelogMergeTreeRewriter
RecordEqualiser valueEqualiser,
boolean changelogRowDeduplicate) {
super(
+ maxLevel,
+ mergeEngine,
readerFactory,
writerFactory,
keyComparator,
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/FullChangelogMergeFunctionWrapper.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/FullChangelogMergeFunctionWrapper.java
index d0de05d40..fe1bf0dc8 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/FullChangelogMergeFunctionWrapper.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/FullChangelogMergeFunctionWrapper.java
@@ -100,11 +100,11 @@ public class FullChangelogMergeFunctionWrapper implements
MergeFunctionWrapper<C
if (isInitialized) {
KeyValue merged = mergeFunction.getResult();
if (topLevelKv == null) {
- if (merged != null && isAdd(merged)) {
+ if (merged != null && merged.isAdd()) {
reusedResult.addChangelog(replace(reusedAfter,
RowKind.INSERT, merged));
}
} else {
- if (merged == null || !isAdd(merged)) {
+ if (merged == null || !merged.isAdd()) {
reusedResult.addChangelog(replace(reusedBefore,
RowKind.DELETE, topLevelKv));
} else if (!changelogRowDeduplicate
|| !valueEqualiser.equals(topLevelKv.value(),
merged.value())) {
@@ -115,7 +115,7 @@ public class FullChangelogMergeFunctionWrapper implements
MergeFunctionWrapper<C
}
return reusedResult.setResultIfNotRetract(merged);
} else {
- if (topLevelKv == null && isAdd(initialKv)) {
+ if (topLevelKv == null && initialKv.isAdd()) {
reusedResult.addChangelog(replace(reusedAfter, RowKind.INSERT,
initialKv));
}
// either topLevelKv is not null, but there is only one kv,
@@ -129,8 +129,4 @@ public class FullChangelogMergeFunctionWrapper implements
MergeFunctionWrapper<C
private KeyValue replace(KeyValue reused, RowKind valueKind, KeyValue
from) {
return reused.replace(from.key(), from.sequenceNumber(), valueKind,
from.value());
}
-
- private boolean isAdd(KeyValue kv) {
- return kv.valueKind() == RowKind.INSERT || kv.valueKind() ==
RowKind.UPDATE_AFTER;
- }
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/FullChangelogMergeTreeCompactRewriter.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/FullChangelogMergeTreeCompactRewriter.java
index 2cc02ec1d..a0f387b24 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/FullChangelogMergeTreeCompactRewriter.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/FullChangelogMergeTreeCompactRewriter.java
@@ -18,6 +18,7 @@
package org.apache.paimon.mergetree.compact;
+import org.apache.paimon.CoreOptions;
import org.apache.paimon.KeyValue;
import org.apache.paimon.codegen.RecordEqualiser;
import org.apache.paimon.data.InternalRow;
@@ -35,10 +36,9 @@ import java.util.List;
/** A {@link MergeTreeCompactRewriter} which produces changelog files for each
full compaction. */
public class FullChangelogMergeTreeCompactRewriter extends
ChangelogMergeTreeRewriter {
- private final int maxLevel;
-
public FullChangelogMergeTreeCompactRewriter(
int maxLevel,
+ CoreOptions.MergeEngine mergeEngine,
KeyValueFileReaderFactory readerFactory,
KeyValueFileWriterFactory writerFactory,
Comparator<InternalRow> keyComparator,
@@ -47,6 +47,8 @@ public class FullChangelogMergeTreeCompactRewriter extends
ChangelogMergeTreeRew
RecordEqualiser valueComparator,
boolean changelogRowDeduplicate) {
super(
+ maxLevel,
+ mergeEngine,
readerFactory,
writerFactory,
keyComparator,
@@ -54,7 +56,6 @@ public class FullChangelogMergeTreeCompactRewriter extends
ChangelogMergeTreeRew
mergeSorter,
valueComparator,
changelogRowDeduplicate);
- this.maxLevel = maxLevel;
}
@Override
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/LookupChangelogMergeFunctionWrapper.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/LookupChangelogMergeFunctionWrapper.java
index 717a32727..bad97e8bb 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/LookupChangelogMergeFunctionWrapper.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/LookupChangelogMergeFunctionWrapper.java
@@ -100,7 +100,6 @@ public class LookupChangelogMergeFunctionWrapper implements
MergeFunctionWrapper
// 2. With level 0, with the latest high level, return changelog
if (highLevel != null) {
- // For first row, we should just return old value. And produce no
changelog.
setChangelog(highLevel, result);
return reusedResult.setResult(result);
}
@@ -121,12 +120,12 @@ public class LookupChangelogMergeFunctionWrapper
implements MergeFunctionWrapper
}
private void setChangelog(KeyValue before, KeyValue after) {
- if (before == null || !isAdd(before)) {
- if (isAdd(after)) {
+ if (before == null || !before.isAdd()) {
+ if (after.isAdd()) {
reusedResult.addChangelog(replaceAfter(RowKind.INSERT, after));
}
} else {
- if (!isAdd(after)) {
+ if (!after.isAdd()) {
reusedResult.addChangelog(replaceBefore(RowKind.DELETE,
before));
} else if (!changelogRowDeduplicate
|| !valueEqualiser.equals(before.value(), after.value())) {
@@ -148,8 +147,4 @@ public class LookupChangelogMergeFunctionWrapper implements
MergeFunctionWrapper
private KeyValue replace(KeyValue reused, RowKind valueKind, KeyValue
from) {
return reused.replace(from.key(), from.sequenceNumber(), valueKind,
from.value());
}
-
- private boolean isAdd(KeyValue kv) {
- return kv.valueKind() == RowKind.INSERT || kv.valueKind() ==
RowKind.UPDATE_AFTER;
- }
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/LookupMergeTreeCompactRewriter.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/LookupMergeTreeCompactRewriter.java
index e5a13b1c6..68df5593d 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/LookupMergeTreeCompactRewriter.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/LookupMergeTreeCompactRewriter.java
@@ -18,6 +18,7 @@
package org.apache.paimon.mergetree.compact;
+import org.apache.paimon.CoreOptions;
import org.apache.paimon.KeyValue;
import org.apache.paimon.codegen.RecordEqualiser;
import org.apache.paimon.data.InternalRow;
@@ -42,6 +43,8 @@ public class LookupMergeTreeCompactRewriter extends
ChangelogMergeTreeRewriter {
private final LookupLevels lookupLevels;
public LookupMergeTreeCompactRewriter(
+ int maxLevel,
+ CoreOptions.MergeEngine mergeEngine,
LookupLevels lookupLevels,
KeyValueFileReaderFactory readerFactory,
KeyValueFileWriterFactory writerFactory,
@@ -51,6 +54,8 @@ public class LookupMergeTreeCompactRewriter extends
ChangelogMergeTreeRewriter {
RecordEqualiser valueEqualiser,
boolean changelogRowDeduplicate) {
super(
+ maxLevel,
+ mergeEngine,
readerFactory,
writerFactory,
keyComparator,
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java
b/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java
index da3667849..3c752472f 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java
@@ -214,10 +214,13 @@ public class KeyValueFileStoreWrite extends
MemoryFileStoreWrite<KeyValue> {
KeyValueFileWriterFactory writerFactory =
writerFactoryBuilder.build(partition, bucket, options);
MergeSorter mergeSorter = new MergeSorter(options, keyType, valueType,
ioManager);
+ int maxLevel = options.numLevels() - 1;
+ CoreOptions.MergeEngine mergeEngine = options.mergeEngine();
switch (options.changelogProducer()) {
case FULL_COMPACTION:
return new FullChangelogMergeTreeCompactRewriter(
- options.numLevels() - 1,
+ maxLevel,
+ mergeEngine,
readerFactory,
writerFactory,
keyComparator,
@@ -226,7 +229,7 @@ public class KeyValueFileStoreWrite extends
MemoryFileStoreWrite<KeyValue> {
valueEqualiserSupplier.get(),
options.changelogRowDeduplicate());
case LOOKUP:
- if (options.mergeEngine() ==
CoreOptions.MergeEngine.FIRST_ROW) {
+ if (mergeEngine == CoreOptions.MergeEngine.FIRST_ROW) {
KeyValueFileReaderFactory keyOnlyReader =
readerFactoryBuilder
.copyWithoutProjection()
@@ -234,6 +237,8 @@ public class KeyValueFileStoreWrite extends
MemoryFileStoreWrite<KeyValue> {
.build(partition, bucket);
ContainsLevels containsLevels =
createContainsLevels(levels, keyOnlyReader);
return new FirstRowMergeTreeCompactRewriter(
+ maxLevel,
+ mergeEngine,
containsLevels,
readerFactory,
writerFactory,
@@ -245,6 +250,8 @@ public class KeyValueFileStoreWrite extends
MemoryFileStoreWrite<KeyValue> {
}
LookupLevels lookupLevels = createLookupLevels(levels,
readerFactory);
return new LookupMergeTreeCompactRewriter(
+ maxLevel,
+ mergeEngine,
lookupLevels,
readerFactory,
writerFactory,
diff --git
a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/LookupCompactionTest.scala
b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/LookupCompactionTest.scala
new file mode 100644
index 000000000..012e69cf1
--- /dev/null
+++
b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/LookupCompactionTest.scala
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.paimon.spark.sql
+
+import org.apache.paimon.CoreOptions
+import org.apache.paimon.fs.{FileStatus, Path}
+import org.apache.paimon.spark.PaimonSparkTestBase
+
+import org.apache.spark.sql.Row
+import org.junit.jupiter.api.Assertions
+
+class LookupCompactionTest extends PaimonSparkTestBase {
+
+ test("Paimon lookup compaction: number of data file written") {
+ CoreOptions.MergeEngine.values().foreach {
+ mergeEngine =>
+ withTable("T") {
+ val extraOptions = if (mergeEngine ==
CoreOptions.MergeEngine.AGGREGATE) {
+ s", 'fields.count.aggregate-function' = 'sum'"
+ } else {
+ ""
+ }
+
+ spark.sql(
+ s"""
+ |CREATE TABLE T (id INT, name STRING, count INT)
+ |TBLPROPERTIES ('primary-key' = 'id', 'merge-engine' =
'$mergeEngine', 'changelog-producer' = 'lookup' $extraOptions)
+ |""".stripMargin)
+
+ val table = loadTable("T")
+ val tabLocation = table.location()
+ val fileIO = table.fileIO()
+
+ // First insert, file is upgraded to the max level when compaction,
no need rewrite
+ spark.sql("INSERT INTO T VALUES (1, 'aaaaaaaaaaa', 1), (2, 'b', 2)")
+ var files = fileIO.listStatus(new Path(tabLocation, "bucket-0"))
+ Assertions.assertEquals(1, dataFileCount(files))
+
+ checkAnswer(
+ spark.sql("SELECT * FROM T ORDER BY id"),
+ Row(1, "aaaaaaaaaaa", 1) :: Row(2, "b", 2) :: Nil)
+
+ spark.sql("INSERT INTO T VALUES (2, 'b', 22), (3, 'c', 3)")
+ files = fileIO.listStatus(new Path(tabLocation, "bucket-0"))
+ // Second insert, file is upgraded to other lower level when
compaction, only DEDUPLICATE can skip rewrite
+ if (mergeEngine == CoreOptions.MergeEngine.DEDUPLICATE) {
+ Assertions.assertEquals(2, dataFileCount(files))
+ } else {
+ Assertions.assertEquals(3, dataFileCount(files))
+ }
+
+ val df = spark.sql("SELECT * FROM T ORDER BY id")
+ mergeEngine match {
+ case CoreOptions.MergeEngine.DEDUPLICATE |
CoreOptions.MergeEngine.PARTIAL_UPDATE =>
+ checkAnswer(df, Row(1, "aaaaaaaaaaa", 1) :: Row(2, "b", 22) ::
Row(3, "c", 3) :: Nil)
+ case CoreOptions.MergeEngine.AGGREGATE =>
+ checkAnswer(df, Row(1, "aaaaaaaaaaa", 1) :: Row(2, "b", 24) ::
Row(3, "c", 3) :: Nil)
+ case CoreOptions.MergeEngine.FIRST_ROW =>
+ checkAnswer(df, Row(1, "aaaaaaaaaaa", 1) :: Row(2, "b", 2) ::
Row(3, "c", 3) :: Nil)
+ case _ =>
+ }
+ }
+ }
+ }
+
+ private def dataFileCount(files: Array[FileStatus]): Int = {
+ files.count(f => f.getPath.getName.startsWith("data"))
+ }
+}
