Skip to content

Commit 355a05b

Browse files
steFaizsteFaiz
committed
clean code
1 parent e804bb2 commit 355a05b

11 files changed

Lines changed: 640 additions & 173 deletions

File tree

‎paimon-common/src/main/java/org/apache/paimon/data/Blob.java‎

Lines changed: 0 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -38,31 +38,6 @@
3838
@Public
3939
public interface Blob {
4040

41-
/**
42-
* The placeholder blob, mainly for blob update in data-evolution. It should never be exposed to
43-
* users.
44-
*/
45-
Blob PLACE_HOLDER =
46-
new Blob() {
47-
@Override
48-
public byte[] toData() {
49-
throw new UnsupportedOperationException(
50-
"Should never call this method for placeholder blob.");
51-
}
52-
53-
@Override
54-
public BlobDescriptor toDescriptor() {
55-
throw new UnsupportedOperationException(
56-
"Should never call this method for placeholder blob.");
57-
}
58-
59-
@Override
60-
public SeekableInputStream newInputStream() throws IOException {
61-
throw new UnsupportedOperationException(
62-
"Should never call this method for placeholder blob.");
63-
}
64-
};
65-
6641
byte[] toData();
6742

6843
BlobDescriptor toDescriptor();

‎paimon-common/src/main/java/org/apache/paimon/data/BlobPlaceholder.java‎

Lines changed: 59 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,59 @@
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.data;
20+
21+
import org.apache.paimon.fs.SeekableInputStream;
22+
23+
import java.io.IOException;
24+
import java.io.Serializable;
25+
26+
/**
27+
* The placeholder blob, mainly for blob update in data-evolution. It should never be exposed to
28+
* users.
29+
*/
30+
public class BlobPlaceholder implements Blob, Serializable {
31+
32+
private static final long serialVersionUID = 1L;
33+
34+
public static final BlobPlaceholder INSTANCE = new BlobPlaceholder();
35+
36+
private BlobPlaceholder() {}
37+
38+
private Object readResolve() {
39+
return INSTANCE;
40+
}
41+
42+
@Override
43+
public byte[] toData() {
44+
throw new UnsupportedOperationException(
45+
"Should never call this method for placeholder blob.");
46+
}
47+
48+
@Override
49+
public BlobDescriptor toDescriptor() {
50+
throw new UnsupportedOperationException(
51+
"Should never call this method for placeholder blob.");
52+
}
53+
54+
@Override
55+
public SeekableInputStream newInputStream() throws IOException {
56+
throw new UnsupportedOperationException(
57+
"Should never call this method for placeholder blob.");
58+
}
59+
}

‎paimon-common/src/main/java/org/apache/paimon/reader/DataEvolutionFileReader.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -66,7 +66,7 @@ public DataEvolutionFileReader(
6666
rowOffsets.length == fieldOffsets.length,
6767
"Row offsets and field offsets must have the same length");
6868
checkArgument(rowOffsets.length > 0, "Row offsets must not be empty");
69-
checkArgument(readers != null && readers.length > 1, "Readers should be more than 1");
69+
checkArgument(readers != null && readers.length >= 1, "should not pass empty readers.");
7070
this.rowOffsets = rowOffsets;
7171
this.fieldOffsets = fieldOffsets;
7272
this.readers = readers;

‎paimon-core/src/main/java/org/apache/paimon/operation/BlobFallbackRecordReader.java‎

Lines changed: 52 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,7 @@
1919
package org.apache.paimon.operation;
2020

2121
import org.apache.paimon.append.ForceSingleBatchReader;
22-
import org.apache.paimon.data.Blob;
22+
import org.apache.paimon.data.BlobPlaceholder;
2323
import org.apache.paimon.data.GenericRow;
2424
import org.apache.paimon.data.InternalRow;
2525
import org.apache.paimon.io.DataFileMeta;
@@ -60,21 +60,23 @@ public class BlobFallbackRecordReader implements RecordReader<InternalRow> {
6060

6161
BlobFallbackRecordReader(
6262
List<DataFileMeta> files,
63-
long rowCount,
6463
BlobFileReaderFactory readerFactory,
6564
List<Range> rowRanges,
6665
RowType readRowType,
6766
int blobIndex) {
6867
this.blobIndex = blobIndex;
6968

7069
checkArgument(!files.isEmpty(), "Blob bunch should not be empty.");
71-
long firstRowId =
72-
files.stream().mapToLong(DataFileMeta::nonNullFirstRowId).min().getAsLong();
73-
long lastRowId = firstRowId + rowCount - 1;
70+
long firstRowId = Long.MAX_VALUE;
71+
long lastRowId = Long.MIN_VALUE;
7472

7573
// sort group readers in descending order
7674
Map<Long, List<DataFileMeta>> sequenceGroups = new TreeMap<>(reverseOrder());
7775
for (DataFileMeta file : files) {
76+
Range fileRange = file.nonNullRowIdRange();
77+
firstRowId = Math.min(firstRowId, fileRange.from);
78+
lastRowId = Math.max(lastRowId, fileRange.to);
79+
7880
sequenceGroups
7981
.computeIfAbsent(file.maxSequenceNumber(), ignored -> new ArrayList<>())
8082
.add(file);
@@ -133,7 +135,12 @@ public RecordIterator<InternalRow> readBatch() throws IOException {
133135
@Override
134136
public InternalRow next() throws IOException {
135137
InternalRow result = null;
136-
// we should always move each iterator forward
138+
// We should always move each iterator forward
139+
// This may significantly increase memory usage and decrease read efficiency
140+
// if `blob-as-descriptor` is disabled and many non-null blobs are updated
141+
// TODO: Do not read stale records if there's a newer non-placeholder
142+
// record. e.g. introduce a discard method to directly discard the
143+
// next record?
137144
for (RecordIterator<InternalRow> iterator : iterators) {
138145
InternalRow row = iterator.next();
139146
if (row == null) {
@@ -161,7 +168,7 @@ public void releaseBatch() {
161168
}
162169

163170
private boolean isPlaceHolder(InternalRow row) {
164-
return !row.isNullAt(blobIndex) && row.getBlob(blobIndex) == Blob.PLACE_HOLDER;
171+
return !row.isNullAt(blobIndex) && row.getBlob(blobIndex) == BlobPlaceholder.INSTANCE;
165172
}
166173

167174
@Override
@@ -190,6 +197,43 @@ public void close() throws IOException {
190197
* will be emitted as placeholder rows.
191198
*
192199
* <p>This reader should always be fully consumed, or the internal states may be broken.
200+
*
201+
* <p>Note that we can not simply concat all data files and read them, even though we guarantee
202+
* writing the full-range data during data-evolution. The complexity is introduced by row-level
203+
* compaction. For example, if we execute following operations:
204+
*
205+
* <ol>
206+
* <li>Write [0, 100], generate files [0, 50], [51, 100]
207+
* <li>Update [0, 100] files, generate files [0, 25], [26, 75], [76, 100]
208+
* <li>Insert new blobs for range [101, 200], generate files [101, 200]
209+
* <li>Update new blobs for range [101, 200], generate files [101, 150], [151, 200]
210+
* <li>Compact, merge [0, 100], [101, 200] to a single range
211+
* <li>Update the compacted files, generate files [0, 200]
212+
* </ol>
213+
*
214+
* <p>The data files layout would be:
215+
*
216+
* <pre>
217+
* |<----------------------- merged range: row 0 ~ 200 --------------------------------->|
218+
* | |
219+
* ┌─────────────────────────┐┌──────────────────────┐
220+
* seq1: │ file1 (0~50) ││ file2 (51~100) │ (empty on 101~200)
221+
* └─────────────────────────┘└──────────────────────┘
222+
* ┌─────────────┐┌──────────────────────┐┌──────────┐
223+
* seq2: │ file3(0~25) ││ file4 (26~75) ││f5(76~100)│ (empty on 101~200)
224+
* └─────────────┘└──────────────────────┘└──────────┘
225+
* ┌────────────────────────────────────┐
226+
* seq3: (empty on 0~100) │ file6 (101~200) │
227+
* └────────────────────────────────────┘
228+
* ┌─────────────────┐┌─────────────────┐
229+
* seq4: (empty on 0~100) │ file7 (101~150) ││ file8 (151~200) │
230+
* └─────────────────┘└─────────────────┘
231+
* ┌───────────────────────────────────────────────────────────────────────────────────────┐
232+
* seq6: │ file9 (0~200) │
233+
* └───────────────────────────────────────────────────────────────────────────────────────┘
234+
* </pre>
235+
*
236+
* <p>We treat all gaps as full-placeholders, and correctly resolve pushed-ranges.
193237
*/
194238
public static class BlobSequenceGroupRecordReader implements RecordReader<InternalRow> {
195239

@@ -327,7 +371,7 @@ public void releaseBatch() {
327371
private InternalRow placeHolderRow() {
328372
if (placeholderRow == null) {
329373
GenericRow row = new GenericRow(readRowType.getFieldCount());
330-
row.setField(blobIndex, Blob.PLACE_HOLDER);
374+
row.setField(blobIndex, BlobPlaceholder.INSTANCE);
331375
placeholderRow = row;
332376
}
333377
return placeholderRow;

‎paimon-core/src/main/java/org/apache/paimon/operation/DataEvolutionSplitRead.java‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -383,7 +383,6 @@ private RecordReader<InternalRow> createFieldBunchReader(
383383
checkArgument(blobIndex >= 0, "Blob bunch read type should contain a blob field.");
384384
return new BlobFallbackRecordReader(
385385
bunch.files(),
386-
bunch.rowCount(),
387386
file ->
388387
createFileReader(
389388
partition,

‎paimon-core/src/test/java/org/apache/paimon/append/BlobTableTest.java‎

Lines changed: 0 additions & 118 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,6 @@
2222
import org.apache.paimon.append.dataevolution.DataEvolutionCompactCoordinator;
2323
import org.apache.paimon.append.dataevolution.DataEvolutionCompactTask;
2424
import org.apache.paimon.catalog.Identifier;
25-
import org.apache.paimon.data.BinaryRow;
2625
import org.apache.paimon.data.BinaryString;
2726
import org.apache.paimon.data.Blob;
2827
import org.apache.paimon.data.BlobData;
@@ -31,22 +30,15 @@
3130
import org.apache.paimon.data.BlobViewStruct;
3231
import org.apache.paimon.data.GenericRow;
3332
import org.apache.paimon.data.InternalRow;
34-
import org.apache.paimon.data.Timestamp;
35-
import org.apache.paimon.format.FormatWriter;
36-
import org.apache.paimon.format.blob.BlobFileFormat;
3733
import org.apache.paimon.fs.FileIO;
3834
import org.apache.paimon.fs.Path;
39-
import org.apache.paimon.fs.PositionOutputStream;
4035
import org.apache.paimon.fs.SeekableInputStream;
4136
import org.apache.paimon.io.DataFileMeta;
42-
import org.apache.paimon.io.DataFilePathFactory;
43-
import org.apache.paimon.manifest.FileSource;
4437
import org.apache.paimon.manifest.ManifestEntry;
4538
import org.apache.paimon.operation.DataEvolutionSplitRead;
4639
import org.apache.paimon.reader.RecordReader;
4740
import org.apache.paimon.schema.Schema;
4841
import org.apache.paimon.schema.SchemaChange;
49-
import org.apache.paimon.stats.SimpleStats;
5042
import org.apache.paimon.table.FileStoreTable;
5143
import org.apache.paimon.table.Table;
5244
import org.apache.paimon.table.TableTestBase;
@@ -56,7 +48,6 @@
5648
import org.apache.paimon.table.sink.CommitMessage;
5749
import org.apache.paimon.table.sink.StreamTableWrite;
5850
import org.apache.paimon.table.sink.StreamWriteBuilder;
59-
import org.apache.paimon.table.source.DataSplit;
6051
import org.apache.paimon.table.source.ReadBuilder;
6152
import org.apache.paimon.table.system.RowTrackingTable;
6253
import org.apache.paimon.types.DataField;
@@ -159,73 +150,6 @@ public void testBasic() throws Exception {
159150
assertThat(integer.get()).isEqualTo(1000);
160151
}
161152

162-
@Test
163-
public void testReadBlobPlaceHolderFallback() throws Exception {
164-
createTableDefault();
165-
writeDataDefault(
166-
Arrays.asList(
167-
GenericRow.of(1, BinaryString.fromString("first"), new BlobData(blobBytes)),
168-
GenericRow.of(
169-
2, BinaryString.fromString("second"), new BlobData(blobBytes)),
170-
GenericRow.of(
171-
3, BinaryString.fromString("third"), new BlobData(blobBytes))));
172-
173-
FileStoreTable table = getTableDefault();
174-
List<DataFileMeta> files =
175-
table.store().newScan().plan().files().stream()
176-
.map(ManifestEntry::file)
177-
.collect(Collectors.toList());
178-
DataFileMeta dataFile =
179-
files.stream()
180-
.filter(file -> !BlobFileFormat.isBlobFile(file.fileName()))
181-
.findFirst()
182-
.get();
183-
DataFileMeta oldBlobFile =
184-
files.stream()
185-
.filter(file -> BlobFileFormat.isBlobFile(file.fileName()))
186-
.findFirst()
187-
.get();
188-
189-
byte[] updatedBytes = "updated-blob".getBytes();
190-
DataFilePathFactory pathFactory =
191-
table.store().pathFactory().createDataFilePathFactory(BinaryRow.EMPTY_ROW, 0);
192-
DataFileMeta newBlobFile =
193-
writeBlobFile(
194-
table.fileIO(),
195-
pathFactory.newBlobPath(),
196-
Arrays.asList(Blob.PLACE_HOLDER, new BlobData(updatedBytes)),
197-
dataFile.nonNullFirstRowId(),
198-
oldBlobFile.maxSequenceNumber() + 1,
199-
oldBlobFile.schemaId(),
200-
oldBlobFile.writeCols());
201-
202-
DataSplit split =
203-
DataSplit.builder()
204-
.withSnapshot(1L)
205-
.withPartition(BinaryRow.EMPTY_ROW)
206-
.withBucket(0)
207-
.withBucketPath(pathFactory.parent().toString())
208-
.withDataFiles(Arrays.asList(dataFile, newBlobFile, oldBlobFile))
209-
.build();
210-
211-
DataEvolutionSplitRead read =
212-
new DataEvolutionSplitRead(
213-
table.fileIO(),
214-
table.schemaManager(),
215-
table.schema(),
216-
table.rowType(),
217-
table.coreOptions(),
218-
table.store().pathFactory());
219-
220-
List<byte[]> actual = new ArrayList<>();
221-
read.createReader(split).forEachRemaining(row -> actual.add(row.getBlob(2).toData()));
222-
223-
assertThat(actual.size()).isEqualTo(3);
224-
assertThat(actual.get(0)).isEqualTo(blobBytes);
225-
assertThat(actual.get(1)).isEqualTo(updatedBytes);
226-
assertThat(actual.get(2)).isEqualTo(blobBytes);
227-
}
228-
229153
@Test
230154
public void testWriteByInputStream() throws Exception {
231155
createTableDefault();
@@ -1047,48 +971,6 @@ private static void writeFile(FileIO fileIO, Path path, byte[] bytes) throws IOE
1047971
}
1048972
}
1049973

1050-
private static DataFileMeta writeBlobFile(
1051-
FileIO fileIO,
1052-
Path path,
1053-
List<Blob> blobs,
1054-
long firstRowId,
1055-
long maxSequenceNumber,
1056-
long schemaId,
1057-
List<String> writeCols)
1058-
throws IOException {
1059-
try (PositionOutputStream out = fileIO.newOutputStream(path, false)) {
1060-
FormatWriter writer =
1061-
new BlobFileFormat()
1062-
.createWriterFactory(RowType.of(DataTypes.BLOB()))
1063-
.create(out, "none");
1064-
for (Blob blob : blobs) {
1065-
writer.addElement(GenericRow.of(blob));
1066-
}
1067-
writer.close();
1068-
}
1069-
return DataFileMeta.create(
1070-
path.getName(),
1071-
fileIO.getFileSize(path),
1072-
blobs.size(),
1073-
DataFileMeta.EMPTY_MIN_KEY,
1074-
DataFileMeta.EMPTY_MAX_KEY,
1075-
SimpleStats.EMPTY_STATS,
1076-
SimpleStats.EMPTY_STATS,
1077-
0,
1078-
maxSequenceNumber,
1079-
schemaId,
1080-
DataFileMeta.DUMMY_LEVEL,
1081-
Collections.emptyList(),
1082-
Timestamp.fromEpochMillis(System.currentTimeMillis()),
1083-
0L,
1084-
null,
1085-
FileSource.APPEND,
1086-
null,
1087-
null,
1088-
firstRowId,
1089-
writeCols);
1090-
}
1091-
1092974
private static long countFilesWithSuffix(FileIO fileIO, Path root, String suffix)
1093975
throws IOException {
1094976
long count = 0;

0 commit comments

Comments
 (0)