Skip to content

Commit

Permalink
[core] Incremental-between tags should deduplicate records (#3338)
Browse files Browse the repository at this point in the history
  • Loading branch information
@JingsongLi
JingsongLi committed May 30, 2024
1 parent b194427 commit de32dae
Show file tree
Hide file tree
Showing 13 changed files with 566 additions and 121 deletions.
4 changes: 4 additions & 0 deletions paimon-core/src/main/java/org/apache/paimon/mergetree/MergeSorter.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -95,6 +95,10 @@ public MemorySegmentPool memoryPool() {
return memoryPool;
}

public RowType valueType() {
return valueType;
}

public void setIOManager(IOManager ioManager) {
this.ioManager = ioManager;
}
Expand Down
109 changes: 41 additions & 68 deletions paimon-core/src/main/java/org/apache/paimon/operation/MergeFileSplitRead.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -116,7 +116,16 @@ public MergeFileSplitRead withKeyProjection(@Nullable int[][] projectedFields) {
return this;
}

public MergeFileSplitRead withValueProjection(@Nullable int[][] projectedFields) {
public Comparator<InternalRow> keyComparator() {
return keyComparator;
}

public MergeSorter mergeSorter() {
return mergeSorter;
}

@Override
public MergeFileSplitRead withProjection(@Nullable int[][] projectedFields) {
if (projectedFields == null) {
return this;
}
Expand Down Expand Up @@ -160,11 +169,13 @@ public MergeFileSplitRead withValueProjection(@Nullable int[][] projectedFields)
return this;
}

@Override
public MergeFileSplitRead withIOManager(IOManager ioManager) {
this.mergeSorter.setIOManager(ioManager);
return this;
}

@Override
public MergeFileSplitRead forceKeepDelete() {
this.forceKeepDelete = true;
return this;
Expand Down Expand Up @@ -211,75 +222,28 @@ public MergeFileSplitRead withFilter(Predicate predicate) {

@Override
public RecordReader<KeyValue> createReader(DataSplit split) throws IOException {
RecordReader<KeyValue> reader = createReaderWithoutOuterProjection(split);
if (outerProjection != null) {
ProjectedRow projectedRow = ProjectedRow.from(outerProjection);
reader = reader.transform(kv -> kv.replaceValue(projectedRow.replaceRow(kv.value())));
if (!split.beforeFiles().isEmpty()) {
throw new IllegalArgumentException("This read cannot accept split with before files.");
}
return reader;
}

private RecordReader<KeyValue> createReaderWithoutOuterProjection(DataSplit split)
throws IOException {
if (split.beforeFiles().isEmpty()) {
if (split.isStreaming() || split.convertToRawFiles().isPresent()) {
return noMergeRead(
split.partition(),
split.bucket(),
split.dataFiles(),
split.deletionFiles().orElse(null),
split.isStreaming());
} else {
return projectKey(
mergeRead(
split.partition(),
split.bucket(),
split.dataFiles(),
null,
forceKeepDelete));
}
} else if (split.isStreaming()) {
// streaming concat read
return ConcatRecordReader.create(
() ->
new ReverseReader(
noMergeRead(
split.partition(),
split.bucket(),
split.beforeFiles(),
split.beforeDeletionFiles().orElse(null),
true)),
() ->
noMergeRead(
split.partition(),
split.bucket(),
split.dataFiles(),
split.deletionFiles().orElse(null),
true));
if (split.isStreaming()) {
return createNoMergeReader(
split.partition(),
split.bucket(),
split.dataFiles(),
split.deletionFiles().orElse(null),
split.isStreaming());
} else {
// batch diff read
return projectKey(
DiffReader.readDiff(
mergeRead(
split.partition(),
split.bucket(),
split.beforeFiles(),
split.beforeDeletionFiles().orElse(null),
false),
mergeRead(
split.partition(),
split.bucket(),
split.dataFiles(),
split.deletionFiles().orElse(null),
false),
keyComparator,
createUdsComparator(),
mergeSorter,
forceKeepDelete));
return createMergeReader(
split.partition(),
split.bucket(),
split.dataFiles(),
split.deletionFiles().orElse(null),
forceKeepDelete);
}
}

private RecordReader<KeyValue> mergeRead(
public RecordReader<KeyValue> createMergeReader(
BinaryRow partition,
int bucket,
List<DataFileMeta> files,
Expand Down Expand Up @@ -316,10 +280,10 @@ private RecordReader<KeyValue> mergeRead(
reader = new DropDeleteReader(reader);
}

return reader;
return projectOuter(projectKey(reader));
}

private RecordReader<KeyValue> noMergeRead(
public RecordReader<KeyValue> createNoMergeReader(
BinaryRow partition,
int bucket,
List<DataFileMeta> files,
Expand All @@ -344,7 +308,8 @@ private RecordReader<KeyValue> noMergeRead(
file.schemaId(), fileName, file.fileSize(), file.level());
});
}
return ConcatRecordReader.create(suppliers);

return projectOuter(ConcatRecordReader.create(suppliers));
}

private Optional<String> changelogFile(DataFileMeta fileMeta) {
Expand All @@ -365,8 +330,16 @@ private RecordReader<KeyValue> projectKey(RecordReader<KeyValue> reader) {
return reader.transform(kv -> kv.replaceKey(projectedRow.replaceRow(kv.key())));
}

private RecordReader<KeyValue> projectOuter(RecordReader<KeyValue> reader) {
if (outerProjection != null) {
ProjectedRow projectedRow = ProjectedRow.from(outerProjection);
reader = reader.transform(kv -> kv.replaceValue(projectedRow.replaceRow(kv.value())));
}
return reader;
}

@Nullable
private UserDefinedSeqComparator createUdsComparator() {
public UserDefinedSeqComparator createUdsComparator() {
return UserDefinedSeqComparator.create(
readerFactoryBuilder.projectedValueType(), sequenceFields);
}
Expand Down
12 changes: 12 additions & 0 deletions paimon-core/src/main/java/org/apache/paimon/operation/RawFileSplitRead.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.deletionvectors.ApplyDeletionVectorReader;
import org.apache.paimon.deletionvectors.DeletionVector;
import org.apache.paimon.disk.IOManager;
import org.apache.paimon.format.FileFormatDiscover;
import org.apache.paimon.format.FormatKey;
import org.apache.paimon.format.FormatReaderContext;
Expand Down Expand Up @@ -98,6 +99,17 @@ public RawFileSplitRead(
this.projection = Projection.range(0, rowType.getFieldCount()).toNestedIndexes();
}

@Override
public SplitRead<InternalRow> forceKeepDelete() {
return this;
}

@Override
public SplitRead<InternalRow> withIOManager(@Nullable IOManager ioManager) {
return this;
}

@Override
public RawFileSplitRead withProjection(int[][] projectedFields) {
if (projectedFields != null) {
projection = projectedFields;
Expand Down
46 changes: 45 additions & 1 deletion paimon-core/src/main/java/org/apache/paimon/operation/SplitRead.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,9 +18,13 @@

package org.apache.paimon.operation;

import org.apache.paimon.disk.IOManager;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.utils.IOFunction;

import javax.annotation.Nullable;

import java.io.IOException;

Expand All @@ -31,8 +35,48 @@
*/
public interface SplitRead<T> {

SplitRead<T> withFilter(Predicate predicate);
SplitRead<T> forceKeepDelete();

SplitRead<T> withIOManager(@Nullable IOManager ioManager);

SplitRead<T> withProjection(@Nullable int[][] projectedFields);

SplitRead<T> withFilter(@Nullable Predicate predicate);

/** Create a {@link RecordReader} from split. */
RecordReader<T> createReader(DataSplit split) throws IOException;

static <L, R> SplitRead<R> convert(
SplitRead<L> read, IOFunction<DataSplit, RecordReader<R>> convertedFactory) {
return new SplitRead<R>() {
@Override
public SplitRead<R> forceKeepDelete() {
read.forceKeepDelete();
return this;
}

@Override
public SplitRead<R> withIOManager(@Nullable IOManager ioManager) {
read.withIOManager(ioManager);
return this;
}

@Override
public SplitRead<R> withProjection(@Nullable int[][] projectedFields) {
read.withProjection(projectedFields);
return this;
}

@Override
public SplitRead<R> withFilter(@Nullable Predicate predicate) {
read.withFilter(predicate);
return this;
}

@Override
public RecordReader<R> createReader(DataSplit split) throws IOException {
return convertedFactory.apply(split);
}
};
}
}
Loading

0 comments on commit de32dae

Please sign in to comment.