You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@iceberg.apache.org by Omar Aloraini <al...@gmail.com> on 2021/02/10 18:04:44 UTC
Iceberg transaction experience
Hello All,
Over the last two months I have been using Iceberg, for the most part it
did what I expected, but when I started using the transaction API
(Table::newTranscation) I came across a few of what I consider
counter-intuitive, at least for my perception of what a transaction is.
My team's goal was to sync data from an existing system(RDBMS) to Hadoop,
and do it reliably, transactions were essential because we wanted to avoid
data duplication, the basic idea was to use the table properties to store a
'marker' indicating how far into the source we have synced. Things were
going well, we had a test that attempted to start two jobs and asserted
that one of them failed.
After sometime that test started failing, Iceberg by default retries the
transaction a few times if there was a conflict error, although that would
make sense for API such as 'newAppend', for newTranasction it didn't, at
least to me. When you start transaction you may query some external state
and then commit the transaction, retrying doesn't align with what I think
most people's experience with SQL transactions.
An easy fix, set 'COMMIT_NUM_RETRIES' to zero.
Days went by, then the same test started failing again, after some time
looking into the source code, I noticed that Iceberg for some operations
validates the commit, but for others it updates the table metadata and
moves forward with the transaction. Best fix I could find was to override
the TableOps, and override the commit method to compare the table
properties at the start and at the commit method.
final record SerializablePropertiesTableOps(
Settings settings,
TableOperations delegate,
MsSqlRowVersion startRv) implements TableOperations {
static Logger LOG =
LogManager.getLogger(SerializablePropertiesTableOps.class);
@Override
public TableMetadata current() {
return delegate.current();
}
@Override
public TableMetadata refresh() {
return delegate.refresh();
}
@Override
public void commit(TableMetadata base, TableMetadata metadata) {
var oldRv =
MsSqlRowVersion.fromBase64(base.properties().get(settings.sync().mssqlMetadataKey()));
var newRv =
MsSqlRowVersion.fromBase64(metadata.properties().get(settings.sync().mssqlMetadataKey()));
LOG.debug("start row versions: {}", startRv);
LOG.debug("old row versions: {}", oldRv);
LOG.debug("new row versions: {}", newRv);
if (!startRv.equals(oldRv)) {
throw new CommitFailedException("table was modified during
transaction, start rv = %s, current rv = %s"
.formatted(startRv, oldRv));
}
delegate.commit(base, metadata);
}
@Override
public FileIO io() {
return delegate.io();
}
@Override
public String metadataFileLocation(String fileName) {
return delegate.metadataFileLocation(fileName);
}
@Override
public LocationProvider locationProvider() {
return delegate.locationProvider();
}
}
Just wanted to share my experience.
Regards
Re: Iceberg transaction experience
Posted by Ryan Blue <rb...@netflix.com.INVALID>.
I think that the only way to do what you want to is to add extra validation
hooks, or to turn off retries like you're doing. If you want to work on
adding validations, I'm happy to work on it with you.
On Wed, Feb 10, 2021 at 12:03 PM Omar Aloraini <al...@gmail.com>
wrote:
> Hi Ryan, thanks for the reply
>
> I have seen a discussion on a pull request that you (and others) have
> suggested using the snapshot summary for the tracking data, this was
> several weeks ago, and I think I tried it, but I didn't work out. If two
> instances of the jobs were running at the same time(by accident) they will
> both attempt to sync data, and they both might succeed, and you will end up
> with duplicated data, correct me If I'm wrong, but after looking at the
> source code, I think two 'newAppend' will commit even if they both started
> from the same TableMetadata(base).
>
> Changes are idempotent, but you can't do: set x to y if x = A, a CAS
> basically.
>
> I remember seeing a 'validate' method which 'newOverwrite' and others
> used, but I couldn't figure out how to use it. My current solution seems to
> work just fine, and if in the future the Transaction interface becomes a
> bit more flexible(adding custom validation hooks), I will use it.
>
>
> On Wed, Feb 10, 2021 at 9:41 PM Ryan Blue <rb...@netflix.com.invalid>
> wrote:
>
>> Thanks, Omar.
>>
>> For use cases like this, I recommend adding the tracking data into the
>> snapshot summary rather than table metadata. That avoids needing to use a
>> transaction and is what we do for similar cases, like streaming sinks that
>> need to ensure checkpoints aren't committed twice.
>>
>> It sounds like the retry behavior is surprising for transactions, but the
>> transaction behaves like any other commit: it retries to avoid failing at a
>> higher level because so much work has already been done by that point. And
>> since that is surprising, changes that are idempotent (set property x to y)
>> don't seem to fit very well.
>>
>> I think what would solve this problem for you is being able to customize
>> validation and add assumptions. That way, you could get a failure if a
>> table property changed from some unexpected value, right? Then you would be
>> able to use Iceberg's retry mechanism but know that your requirement is
>> always met. Does that sound like a solution that would work for you? If so,
>> we can work on adding something like that.
>>
>> rb
>>
>> On Wed, Feb 10, 2021 at 10:05 AM Omar Aloraini <al...@gmail.com>
>> wrote:
>>
>>> Hello All,
>>>
>>> Over the last two months I have been using Iceberg, for the most part it
>>> did what I expected, but when I started using the transaction API
>>> (Table::newTranscation) I came across a few of what I consider
>>> counter-intuitive, at least for my perception of what a transaction is.
>>>
>>> My team's goal was to sync data from an existing system(RDBMS) to
>>> Hadoop, and do it reliably, transactions were essential because we wanted
>>> to avoid data duplication, the basic idea was to use the table properties
>>> to store a 'marker' indicating how far into the source we have synced.
>>> Things were going well, we had a test that attempted to start two jobs and
>>> asserted that one of them failed.
>>>
>>> After sometime that test started failing, Iceberg by default retries the
>>> transaction a few times if there was a conflict error, although that would
>>> make sense for API such as 'newAppend', for newTranasction it didn't, at
>>> least to me. When you start transaction you may query some external state
>>> and then commit the transaction, retrying doesn't align with what I think
>>> most people's experience with SQL transactions.
>>>
>>> An easy fix, set 'COMMIT_NUM_RETRIES' to zero.
>>>
>>> Days went by, then the same test started failing again, after some time
>>> looking into the source code, I noticed that Iceberg for some operations
>>> validates the commit, but for others it updates the table metadata and
>>> moves forward with the transaction. Best fix I could find was to override
>>> the TableOps, and override the commit method to compare the table
>>> properties at the start and at the commit method.
>>> final record SerializablePropertiesTableOps(
>>> Settings settings,
>>> TableOperations delegate,
>>> MsSqlRowVersion startRv) implements TableOperations {
>>>
>>> static Logger LOG =
>>> LogManager.getLogger(SerializablePropertiesTableOps.class);
>>>
>>> @Override
>>> public TableMetadata current() {
>>> return delegate.current();
>>> }
>>>
>>> @Override
>>> public TableMetadata refresh() {
>>> return delegate.refresh();
>>> }
>>>
>>> @Override
>>> public void commit(TableMetadata base, TableMetadata metadata) {
>>> var oldRv =
>>> MsSqlRowVersion.fromBase64(base.properties().get(settings.sync().mssqlMetadataKey()));
>>> var newRv =
>>> MsSqlRowVersion.fromBase64(metadata.properties().get(settings.sync().mssqlMetadataKey()));
>>>
>>> LOG.debug("start row versions: {}", startRv);
>>> LOG.debug("old row versions: {}", oldRv);
>>> LOG.debug("new row versions: {}", newRv);
>>>
>>> if (!startRv.equals(oldRv)) {
>>> throw new CommitFailedException("table was modified during
>>> transaction, start rv = %s, current rv = %s"
>>> .formatted(startRv, oldRv));
>>> }
>>>
>>> delegate.commit(base, metadata);
>>> }
>>>
>>> @Override
>>> public FileIO io() {
>>> return delegate.io();
>>> }
>>>
>>> @Override
>>> public String metadataFileLocation(String fileName) {
>>> return delegate.metadataFileLocation(fileName);
>>> }
>>>
>>> @Override
>>> public LocationProvider locationProvider() {
>>> return delegate.locationProvider();
>>> }
>>> }
>>>
>>> Just wanted to share my experience.
>>>
>>> Regards
>>>
>>
>>
>> --
>> Ryan Blue
>> Software Engineer
>> Netflix
>>
>
--
Ryan Blue
Software Engineer
Netflix
Re: Iceberg transaction experience
Posted by Omar Aloraini <al...@gmail.com>.
Hi Ryan, thanks for the reply
I have seen a discussion on a pull request that you (and others) have
suggested using the snapshot summary for the tracking data, this was
several weeks ago, and I think I tried it, but I didn't work out. If two
instances of the jobs were running at the same time(by accident) they will
both attempt to sync data, and they both might succeed, and you will end up
with duplicated data, correct me If I'm wrong, but after looking at the
source code, I think two 'newAppend' will commit even if they both started
from the same TableMetadata(base).
Changes are idempotent, but you can't do: set x to y if x = A, a CAS
basically.
I remember seeing a 'validate' method which 'newOverwrite' and others
used, but I couldn't figure out how to use it. My current solution seems to
work just fine, and if in the future the Transaction interface becomes a
bit more flexible(adding custom validation hooks), I will use it.
On Wed, Feb 10, 2021 at 9:41 PM Ryan Blue <rb...@netflix.com.invalid> wrote:
> Thanks, Omar.
>
> For use cases like this, I recommend adding the tracking data into the
> snapshot summary rather than table metadata. That avoids needing to use a
> transaction and is what we do for similar cases, like streaming sinks that
> need to ensure checkpoints aren't committed twice.
>
> It sounds like the retry behavior is surprising for transactions, but the
> transaction behaves like any other commit: it retries to avoid failing at a
> higher level because so much work has already been done by that point. And
> since that is surprising, changes that are idempotent (set property x to y)
> don't seem to fit very well.
>
> I think what would solve this problem for you is being able to customize
> validation and add assumptions. That way, you could get a failure if a
> table property changed from some unexpected value, right? Then you would be
> able to use Iceberg's retry mechanism but know that your requirement is
> always met. Does that sound like a solution that would work for you? If so,
> we can work on adding something like that.
>
> rb
>
> On Wed, Feb 10, 2021 at 10:05 AM Omar Aloraini <al...@gmail.com>
> wrote:
>
>> Hello All,
>>
>> Over the last two months I have been using Iceberg, for the most part it
>> did what I expected, but when I started using the transaction API
>> (Table::newTranscation) I came across a few of what I consider
>> counter-intuitive, at least for my perception of what a transaction is.
>>
>> My team's goal was to sync data from an existing system(RDBMS) to Hadoop,
>> and do it reliably, transactions were essential because we wanted to avoid
>> data duplication, the basic idea was to use the table properties to store a
>> 'marker' indicating how far into the source we have synced. Things were
>> going well, we had a test that attempted to start two jobs and asserted
>> that one of them failed.
>>
>> After sometime that test started failing, Iceberg by default retries the
>> transaction a few times if there was a conflict error, although that would
>> make sense for API such as 'newAppend', for newTranasction it didn't, at
>> least to me. When you start transaction you may query some external state
>> and then commit the transaction, retrying doesn't align with what I think
>> most people's experience with SQL transactions.
>>
>> An easy fix, set 'COMMIT_NUM_RETRIES' to zero.
>>
>> Days went by, then the same test started failing again, after some time
>> looking into the source code, I noticed that Iceberg for some operations
>> validates the commit, but for others it updates the table metadata and
>> moves forward with the transaction. Best fix I could find was to override
>> the TableOps, and override the commit method to compare the table
>> properties at the start and at the commit method.
>> final record SerializablePropertiesTableOps(
>> Settings settings,
>> TableOperations delegate,
>> MsSqlRowVersion startRv) implements TableOperations {
>>
>> static Logger LOG =
>> LogManager.getLogger(SerializablePropertiesTableOps.class);
>>
>> @Override
>> public TableMetadata current() {
>> return delegate.current();
>> }
>>
>> @Override
>> public TableMetadata refresh() {
>> return delegate.refresh();
>> }
>>
>> @Override
>> public void commit(TableMetadata base, TableMetadata metadata) {
>> var oldRv =
>> MsSqlRowVersion.fromBase64(base.properties().get(settings.sync().mssqlMetadataKey()));
>> var newRv =
>> MsSqlRowVersion.fromBase64(metadata.properties().get(settings.sync().mssqlMetadataKey()));
>>
>> LOG.debug("start row versions: {}", startRv);
>> LOG.debug("old row versions: {}", oldRv);
>> LOG.debug("new row versions: {}", newRv);
>>
>> if (!startRv.equals(oldRv)) {
>> throw new CommitFailedException("table was modified during
>> transaction, start rv = %s, current rv = %s"
>> .formatted(startRv, oldRv));
>> }
>>
>> delegate.commit(base, metadata);
>> }
>>
>> @Override
>> public FileIO io() {
>> return delegate.io();
>> }
>>
>> @Override
>> public String metadataFileLocation(String fileName) {
>> return delegate.metadataFileLocation(fileName);
>> }
>>
>> @Override
>> public LocationProvider locationProvider() {
>> return delegate.locationProvider();
>> }
>> }
>>
>> Just wanted to share my experience.
>>
>> Regards
>>
>
>
> --
> Ryan Blue
> Software Engineer
> Netflix
>
Re: Iceberg transaction experience
Posted by Ryan Blue <rb...@netflix.com.INVALID>.
Thanks, Omar.
For use cases like this, I recommend adding the tracking data into the
snapshot summary rather than table metadata. That avoids needing to use a
transaction and is what we do for similar cases, like streaming sinks that
need to ensure checkpoints aren't committed twice.
It sounds like the retry behavior is surprising for transactions, but the
transaction behaves like any other commit: it retries to avoid failing at a
higher level because so much work has already been done by that point. And
since that is surprising, changes that are idempotent (set property x to y)
don't seem to fit very well.
I think what would solve this problem for you is being able to customize
validation and add assumptions. That way, you could get a failure if a
table property changed from some unexpected value, right? Then you would be
able to use Iceberg's retry mechanism but know that your requirement is
always met. Does that sound like a solution that would work for you? If so,
we can work on adding something like that.
rb
On Wed, Feb 10, 2021 at 10:05 AM Omar Aloraini <al...@gmail.com>
wrote:
> Hello All,
>
> Over the last two months I have been using Iceberg, for the most part it
> did what I expected, but when I started using the transaction API
> (Table::newTranscation) I came across a few of what I consider
> counter-intuitive, at least for my perception of what a transaction is.
>
> My team's goal was to sync data from an existing system(RDBMS) to Hadoop,
> and do it reliably, transactions were essential because we wanted to avoid
> data duplication, the basic idea was to use the table properties to store a
> 'marker' indicating how far into the source we have synced. Things were
> going well, we had a test that attempted to start two jobs and asserted
> that one of them failed.
>
> After sometime that test started failing, Iceberg by default retries the
> transaction a few times if there was a conflict error, although that would
> make sense for API such as 'newAppend', for newTranasction it didn't, at
> least to me. When you start transaction you may query some external state
> and then commit the transaction, retrying doesn't align with what I think
> most people's experience with SQL transactions.
>
> An easy fix, set 'COMMIT_NUM_RETRIES' to zero.
>
> Days went by, then the same test started failing again, after some time
> looking into the source code, I noticed that Iceberg for some operations
> validates the commit, but for others it updates the table metadata and
> moves forward with the transaction. Best fix I could find was to override
> the TableOps, and override the commit method to compare the table
> properties at the start and at the commit method.
> final record SerializablePropertiesTableOps(
> Settings settings,
> TableOperations delegate,
> MsSqlRowVersion startRv) implements TableOperations {
>
> static Logger LOG =
> LogManager.getLogger(SerializablePropertiesTableOps.class);
>
> @Override
> public TableMetadata current() {
> return delegate.current();
> }
>
> @Override
> public TableMetadata refresh() {
> return delegate.refresh();
> }
>
> @Override
> public void commit(TableMetadata base, TableMetadata metadata) {
> var oldRv =
> MsSqlRowVersion.fromBase64(base.properties().get(settings.sync().mssqlMetadataKey()));
> var newRv =
> MsSqlRowVersion.fromBase64(metadata.properties().get(settings.sync().mssqlMetadataKey()));
>
> LOG.debug("start row versions: {}", startRv);
> LOG.debug("old row versions: {}", oldRv);
> LOG.debug("new row versions: {}", newRv);
>
> if (!startRv.equals(oldRv)) {
> throw new CommitFailedException("table was modified during
> transaction, start rv = %s, current rv = %s"
> .formatted(startRv, oldRv));
> }
>
> delegate.commit(base, metadata);
> }
>
> @Override
> public FileIO io() {
> return delegate.io();
> }
>
> @Override
> public String metadataFileLocation(String fileName) {
> return delegate.metadataFileLocation(fileName);
> }
>
> @Override
> public LocationProvider locationProvider() {
> return delegate.locationProvider();
> }
> }
>
> Just wanted to share my experience.
>
> Regards
>
--
Ryan Blue
Software Engineer
Netflix