Skip to content

[WIP] Add sequence number for supporting row level delete #588

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
chenjunjiedada wants to merge 14 commits into from
Closed

[WIP] Add sequence number for supporting row level delete #588

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
14 commits
Select commit Hold shift + click to select a range
d29358a
Add sequence number for snapshot
chenjunjiedada Nov 18, 2019
ffc849e
Add more unit tests
chenjunjiedada Dec 3, 2019
09eb9a8
Add sequence number for data file and manifest file
chenjunjiedada Feb 4, 2020
8f1fd1a
Merge branch 'master' into seq
chenjunjiedada Feb 5, 2020
82420a2
add unit test for cherry-pick feature
chenjunjiedada Feb 5, 2020
9ac1143
update unit tests
chenjunjiedada Feb 6, 2020
7313947
some minor changes
chenjunjiedada Feb 8, 2020
e72cc42
fix sequence number genration for multi-thread situation
chenjunjiedada Feb 12, 2020
f7fd87d
Merge branch 'master' into seq
chenjunjiedada Feb 12, 2020
75fa8a8
bump the table format version
chenjunjiedada Feb 17, 2020
8ce5e75
update spec
chenjunjiedada Feb 17, 2020
baeffec
fix words
chenjunjiedada Feb 18, 2020
f38052f
Merge branch 'master' into seq
chenjunjiedada Mar 29, 2020
e72ce7c
address comments
chenjunjiedada Apr 7, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 10 additions & 2 deletions api/src/main/java/org/apache/iceberg/DataFile.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,8 +59,9 @@ static StructType getType(StructType partitionType) {
optional(128, "upper_bounds", MapType.ofRequired(129, 130,
IntegerType.get(), BinaryType.get())),
optional(131, "key_metadata", BinaryType.get()),
optional(132, "split_offsets", ListType.ofRequired(133, LongType.get()))
// NEXT ID TO ASSIGN: 134
optional(132, "split_offsets", ListType.ofRequired(133, LongType.get())),
optional(134, "sequence_number", LongType.get())
// NEXT ID TO ASSIGN: 135
);
}

Expand Down Expand Up @@ -152,4 +153,11 @@ static StructType getType(StructType partitionType) {
* are determined by these offsets. The returned list must be sorted in ascending order.
*/
List<Long> splitOffsets();

/**
* @return The sequence number to identify the order in which data files and deletion files are to be processed.
* If the sequence number is not specified it is inherited from the manifest file struct in the manifest list file.
*/
long sequenceNumber();

}
11 changes: 10 additions & 1 deletion api/src/main/java/org/apache/iceberg/ManifestFile.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,9 @@ public interface ManifestFile {
))),
optional(512, "added_rows_count", Types.LongType.get()),
optional(513, "existing_rows_count", Types.LongType.get()),
optional(514, "deleted_rows_count", Types.LongType.get()));
optional(514, "deleted_rows_count", Types.LongType.get()),
optional(515, "sequence_number", Types.LongType.get())
);

static Schema schema() {
return SCHEMA;
Expand Down Expand Up @@ -128,6 +130,13 @@ default boolean hasDeletedFiles() {
*/
Long deletedRowsCount();

/**
* @return the sequence number of this manifest. The sequence number of manifest stores in manifest list file. Since
* The data files' sequence number is optional, it should inherit the manifest's sequence number if the reader reads
* null from manifest file.
*/
Long sequenceNumber();
Copy link
Contributor

@prodeezy prodeezy Apr 6, 2020 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

So if a seq number is available at manifest file level then we assume all datafiles in that manifestfile have the same seq number. But if there are datafile level seq numbers then should we be tracking any metadata at ManifestFile level? Like Min/Max seq number?

Copy link
Collaborator Author

@chenjunjiedada chenjunjiedada Apr 7, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Currently, the logic is just the first step that if a seq number is not found in data_file, then we try to find it from manifestFile. It doesn't contain the logic about computing the min/max value. I would like to use separated fields to track the min/max sequence number in the next step.

Copy link
Contributor

@prodeezy prodeezy Apr 7, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ok


/**
* Returns a list of {@link PartitionFieldSummary partition field summaries}.
* <p>
Expand Down
7 changes: 7 additions & 0 deletions api/src/main/java/org/apache/iceberg/Snapshot.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -104,4 +104,11 @@ public interface Snapshot {
* @return the location of the manifest list for this Snapshot
*/
String manifestListLocation();

/**
* Return this snapshot's sequence number, or 0 if the table has no snapshot yet.
Copy link
Contributor

@aokolnychyi aokolnychyi Nov 25, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In which case will this actually return 0? If there is no snapshot, then there is no Snapshot object, right?

Copy link
Contributor

@rdblue rdblue Feb 7, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It would make sense to return 0 if the snapshot was written without a sequence number, and to use a primitive, right?

*
* @return the sequence number of this Snapshot
*/
Long sequenceNumber();
}
10 changes: 10 additions & 0 deletions api/src/test/java/org/apache/iceberg/TestHelpers.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -255,6 +255,11 @@ public Long deletedRowsCount() {
return deletedRows;
}

@Override
public Long sequenceNumber() {
return null;
}

@Override
public List<PartitionFieldSummary> partitions() {
return partitions;
Expand Down Expand Up @@ -372,5 +377,10 @@ public DataFile copyWithoutStats() {
public List<Long> splitOffsets() {
return null;
}

@Override
public long sequenceNumber() {
return 0;
}
}
}
6 changes: 4 additions & 2 deletions core/src/main/java/org/apache/iceberg/BaseRewriteManifests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -143,6 +143,7 @@ public RewriteManifests addManifest(ManifestFile manifest) {
Preconditions.checkArgument(
manifest.snapshotId() == null || manifest.snapshotId() == -1,
"Snapshot id must be assigned during commit");
Preconditions.checkArgument(manifest.sequenceNumber() == null, "Sequence number must be assigned during commit");

if (snapshotIdInheritanceEnabled && manifest.snapshotId() == null) {
addedManifests.add(manifest);
Expand Down Expand Up @@ -179,10 +180,11 @@ public List<ManifestFile> apply(TableMetadata base) {

validateFilesCounts();

// TODO: add sequence numbers here
Iterable<ManifestFile> newManifestsWithMetadata = Iterables.transform(
Iterables.concat(newManifests, addedManifests, rewrittenAddedManifests),
manifest -> GenericManifestFile.copyOf(manifest).withSnapshotId(snapshotId()).build());
manifest -> GenericManifestFile.copyOf(manifest).withSnapshotId(snapshotId())
.withSequenceNumber(sequenceNumber())
.build());

// put new manifests at the beginning
List<ManifestFile> apply = new ArrayList<>();
Expand Down
36 changes: 36 additions & 0 deletions core/src/main/java/org/apache/iceberg/BaseSnapshot.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,7 @@ class BaseSnapshot implements Snapshot {
private final InputFile manifestList;
private final String operation;
private final Map<String, String> summary;
private Long sequenceNumber;
Copy link
Contributor

@prodeezy prodeezy Apr 6, 2020 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@chenjunjiedada Can you comment on why the max seq number is not being tracked in Snapshot but instead Snapshot Seq Number is? the Merge-on-Read doc says we use max seq number in the snapshot. Maybe i missed the context. what's the reason for switching to snapshot sequence number?

Copy link
Collaborator Author

@chenjunjiedada chenjunjiedada Apr 7, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@prodeezy , you may want to check this: #588 (comment).


// lazily initialized
private List<ManifestFile> manifests = null;
Expand Down Expand Up @@ -75,6 +76,31 @@ class BaseSnapshot implements Snapshot {
this.manifestList = manifestList;
}

BaseSnapshot(FileIO io,
long snapshotId,
Long parentId,
long timestampMillis,
String operation,
Map<String, String> summary,
InputFile manifestList,
Long sequenceNumber) {
this(io, snapshotId, parentId, timestampMillis, operation, summary, manifestList);
this.sequenceNumber = sequenceNumber;
}

BaseSnapshot(FileIO io,
long snapshotId,
Long parentId,
long timestampMillis,
String operation,
Map<String, String> summary,
List<ManifestFile> manifests,
Long sequenceNumber) {
this(io, snapshotId, parentId, timestampMillis, operation, summary, (InputFile) null);
this.manifests = manifests;
this.sequenceNumber = sequenceNumber;
}

BaseSnapshot(FileIO io,
long snapshotId,
Long parentId,
Expand Down Expand Up @@ -186,6 +212,15 @@ private void cacheChanges() {
this.cachedDeletes = deletes.build();
}

@Override
public Long sequenceNumber() {
Copy link
Contributor

@aokolnychyi aokolnychyi Nov 25, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hm, does this mean that all existing snapshots written without sequence numbers will have sequenceNumber as 0?

Copy link
Collaborator Author

@chenjunjiedada chenjunjiedada Nov 25, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

That's an initial implementation that needs some discussion. For example, we can generate sequence numbers from the timestamp for existing snapshots.

Copy link
Contributor

@aokolnychyi aokolnychyi Nov 26, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think the timestamp-based approach would work for existing snapshots, but I am not sure how we will propagate this info down to manifests and files. For example, ManifestFile keeps track of the snapshot in which the manifest was added, but that snapshot can be already expired, so we won't be able to fetch its metadata. Maybe, defaulting to 0 won't be an issue, let me think more about this.

@rdblue, any thoughts on keeping backward compatibility?

Copy link
Contributor

@rdblue rdblue Feb 7, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

When we start using sequence numbers, anything without a sequence number should be considered older and should default to 0. That will work as long as there is some time at which we always use sequence numbers.

What wouldn't work is if we have some writers that use sequence numbers and some writers that do not because the ones that don't write sequence numbers would be sequenced before writes that do: if some writer creates seq 1 and an older writer commits later, the older writer's new commit would have seq 0. That new writer would probably also drop all sequence numbers from metadata -- the table sequence number and the one for each ManifestFile stored in the ManifestList.

One more thing to note: the older writer would reset sequence numbers, but metadata written by newer clients could still contain explicit sequence numbers. For example, a new writer that performs a compaction with older data will write the older data's sequence numbers into metadata. If the sequence number gets reset to 0 by an older writer after that, then sequence numbers get all mixed up. So we will need to ensure that older writers cannot operate on tables once they move to sequence numbers for correctness. That requires bumping the format version.

Does that reasoning sound correct, @aokolnychyi, @chenjunjiedada?

Copy link
Collaborator Author

@chenjunjiedada chenjunjiedada Feb 8, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, that makes sense. Bumping the format version can prevent an old writer from writing data to a table with a new version format. We may still need to consider how to handle the existing data.

So the question is do we want to allow writing from the old writer and new writer at the same time. Maybe a spark and a presto are allowed to write to the table at the same time? If we don't want to support the old writer and new writer at the same time, we can update the format version for existing data when committing.

Copy link
Contributor

@aokolnychyi aokolnychyi Feb 11, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I agree with Ryan that old writers should not be allowed to write to tables with sequence numbers. It seems like the right time to bump the format version and include new manifest types and snapshot id inheritance. Then we can safely assign 0 sequence number to all files that were written before we start assigning new ones.

if (sequenceNumber == null) {
return 0L;
}

return sequenceNumber;
}

@Override
public String toString() {
return MoreObjects.toStringHelper(this)
Expand All @@ -194,6 +229,7 @@ public String toString() {
.add("operation", operation)
.add("summary", summary)
.add("manifests", manifests())
.add("sequence_number", sequenceNumber())
.toString();
}
}
19 changes: 14 additions & 5 deletions core/src/main/java/org/apache/iceberg/FastAppend.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -95,6 +95,7 @@ public FastAppend appendManifest(ManifestFile manifest) {
Preconditions.checkArgument(
manifest.snapshotId() == null || manifest.snapshotId() == -1,
"Snapshot id must be assigned during commit");
Preconditions.checkArgument(manifest.sequenceNumber() == null, "Sequence number must be assigned during commit");

if (snapshotIdInheritanceEnabled && manifest.snapshotId() == null) {
summaryBuilder.addedManifest(manifest);
Expand Down Expand Up @@ -130,17 +131,25 @@ public List<ManifestFile> apply(TableMetadata base) {
throw new RuntimeIOException(e, "Failed to write manifest");
}

// TODO: add sequence numbers here
Iterable<ManifestFile> newManifestsWithMetadata = Iterables.transform(newManifests,
manifest -> GenericManifestFile.copyOf(manifest).withSequenceNumber(sequenceNumber()).build());

Iterable<ManifestFile> appendManifestsWithMetadata = Iterables.transform(
Iterables.concat(appendManifests, rewrittenAppendManifests),
manifest -> GenericManifestFile.copyOf(manifest).withSnapshotId(snapshotId()).build());
Iterables.addAll(newManifests, appendManifestsWithMetadata);
manifest -> GenericManifestFile.copyOf(manifest).withSnapshotId(snapshotId())
.withSequenceNumber(sequenceNumber())
.build());

List<ManifestFile> manifestsWithMetadata = Lists.newArrayList();
Iterables.addAll(manifestsWithMetadata, newManifestsWithMetadata);
Iterables.addAll(manifestsWithMetadata, appendManifestsWithMetadata);


if (base.currentSnapshot() != null) {
newManifests.addAll(base.currentSnapshot().manifests());
manifestsWithMetadata.addAll(base.currentSnapshot().manifests());
}

return newManifests;
return manifestsWithMetadata;
}

@Override
Expand Down
15 changes: 14 additions & 1 deletion core/src/main/java/org/apache/iceberg/GenericDataFile.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,6 +65,7 @@ public PartitionData copy() {
private Map<Integer, ByteBuffer> upperBounds = null;
private List<Long> splitOffsets = null;
private byte[] keyMetadata = null;
private Long sequenceNumber = null;

// cached schema
private transient org.apache.avro.Schema avroSchema = null;
Expand Down Expand Up @@ -175,6 +176,7 @@ private GenericDataFile(GenericDataFile toCopy, boolean fullCopy) {
this.fileSizeInBytes = toCopy.fileSizeInBytes;
this.fileOrdinal = toCopy.fileOrdinal;
this.sortColumns = copy(toCopy.sortColumns);
this.sequenceNumber = toCopy.sequenceNumber;
if (fullCopy) {
// TODO: support lazy conversion to/from map
this.columnSizes = copy(toCopy.columnSizes);
Expand Down Expand Up @@ -270,6 +272,11 @@ public List<Long> splitOffsets() {
return splitOffsets;
}

@Override
public long sequenceNumber() {
return sequenceNumber == null ? 0 : sequenceNumber;
}

@Override
public org.apache.avro.Schema getSchema() {
if (avroSchema == null) {
Expand Down Expand Up @@ -332,6 +339,9 @@ public void put(int i, Object v) {
case 14:
this.splitOffsets = (List<Long>) v;
return;
case 15:
this.sequenceNumber = (Long) v;
return;
default:
// ignore the object, it must be from a newer version of the format
}
Expand Down Expand Up @@ -382,6 +392,8 @@ public Object get(int i) {
return keyMetadata();
case 14:
return splitOffsets;
case 15:
return sequenceNumber;
default:
throw new UnsupportedOperationException("Unknown field ordinal: " + pos);
}
Expand All @@ -401,7 +413,7 @@ private static org.apache.avro.Schema getAvroSchema(Types.StructType partitionTy

@Override
public int size() {
return 15;
return 16;
}

@Override
Copy link
Member

@openinx openinx Mar 30, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please add the sequence_number inside the toString() ?

Copy link
Member

@openinx openinx Mar 30, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The GenericDataFile(GenericDataFile toCopy, boolean fullCopy) also need to copy the sequence_number I guess.

Expand All @@ -410,6 +422,7 @@ public String toString() {
.add("file_path", filePath)
.add("file_format", format)
.add("partition", partitionData)
.add("sequence_number", sequenceNumber())
.add("record_count", recordCount)
.add("file_size_in_bytes", fileSizeInBytes)
.add("column_sizes", columnSizes)
Expand Down
18 changes: 18 additions & 0 deletions core/src/main/java/org/apache/iceberg/GenericManifestFile.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,7 @@ public class GenericManifestFile
private Integer deletedFilesCount = null;
private Long deletedRowsCount = null;
private List<PartitionFieldSummary> partitions = null;
private Long sequenceNumber = null;
Copy link
Member

@openinx openinx Mar 30, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please also check the similar things in GenericDataFile, such as the constructor ? toString() ? etc.

Copy link
Collaborator Author

@chenjunjiedada chenjunjiedada Apr 7, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

OK


/**
* Used by Avro reflection to instantiate this class when reading manifest files.
Expand Down Expand Up @@ -156,6 +157,7 @@ private GenericManifestFile(GenericManifestFile toCopy) {
this.deletedRowsCount = toCopy.deletedRowsCount;
this.partitions = copyList(toCopy.partitions, PartitionFieldSummary::copy);
this.fromProjectionPos = toCopy.fromProjectionPos;
this.sequenceNumber = toCopy.sequenceNumber;
}

/**
Expand Down Expand Up @@ -227,6 +229,11 @@ public Long deletedRowsCount() {
return deletedRowsCount;
}

@Override
public Long sequenceNumber() {
return sequenceNumber;
}

@Override
public List<PartitionFieldSummary> partitions() {
return partitions;
Expand Down Expand Up @@ -272,6 +279,8 @@ public Object get(int i) {
return existingRowsCount;
case 10:
return deletedRowsCount;
case 11:
return sequenceNumber;
default:
throw new UnsupportedOperationException("Unknown field ordinal: " + pos);
}
Expand Down Expand Up @@ -320,6 +329,9 @@ public <T> void set(int i, T value) {
case 10:
this.deletedRowsCount = (Long) value;
return;
case 11:
this.sequenceNumber = (Long) value;
return;
default:
// ignore the object, it must be from a newer version of the format
}
Expand Down Expand Up @@ -361,6 +373,7 @@ public String toString() {
return MoreObjects.toStringHelper(this)
.add("path", manifestPath)
.add("length", length)
.add("sequence_number", sequenceNumber())
.add("partition_spec_id", specId)
.add("added_snapshot_id", snapshotId)
.add("added_data_files_count", addedFilesCount)
Expand Down Expand Up @@ -397,6 +410,11 @@ public CopyBuilder withSnapshotId(Long newSnapshotId) {
return this;
}

public CopyBuilder withSequenceNumber(Long newSequenceNumber) {
manifestFile.sequenceNumber = newSequenceNumber;
return this;
}

public ManifestFile build() {
return manifestFile;
}
Expand Down
9 changes: 7 additions & 2 deletions core/src/main/java/org/apache/iceberg/InheritableMetadataFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,22 +30,27 @@ static InheritableMetadata empty() {
}

static InheritableMetadata fromManifest(ManifestFile manifest) {
return new BaseInheritableMetadata(manifest.snapshotId());
return new BaseInheritableMetadata(manifest.snapshotId(), manifest.sequenceNumber());
}

static class BaseInheritableMetadata implements InheritableMetadata {

private final Long snapshotId;
private final Long sequenceNumber;

private BaseInheritableMetadata(Long snapshotId) {
private BaseInheritableMetadata(Long snapshotId, Long sequenceNumber) {
this.snapshotId = snapshotId;
this.sequenceNumber = sequenceNumber;
}

@Override
public ManifestEntry apply(ManifestEntry manifestEntry) {
if (manifestEntry.snapshotId() == null) {
manifestEntry.setSnapshotId(snapshotId);
}
if (manifestEntry.sequenceNumber() == null && this.sequenceNumber != null) {
manifestEntry.setSequenceNumber(sequenceNumber);
}
return manifestEntry;
}
}
Expand Down
Loading