You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "ahmedabu98 (via GitHub)" <gi...@apache.org> on 2024/10/18 23:23:25 UTC
[PR] [Managed Iceberg] Allowed updating partition specs at runtime [beam]
ahmedabu98 opened a new pull request, #32879:
URL: https://github.com/apache/beam/pull/32879
Users may update the table's spec while a write pipeline is running (e.g. streaming). Sometimes, this update can happen after serializing DataFiles. The partition spec is not serializable so we don't preserve it, but we do keep its ID. We can use this information to fetch the correct partition spec and recreate the DataFile before appending to the table.
Fixes #32862
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Managed Iceberg] Allow updating partition specs at runtime [beam]
Posted by "DanielMorales9 (via GitHub)" <gi...@apache.org>.
DanielMorales9 commented on code in PR #32879:
URL: https://github.com/apache/beam/pull/32879#discussion_r1807930482
##########
sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/RecordWriterManagerTest.java:
##########
@@ -269,6 +271,25 @@ public void testRequireClosingBeforeFetchingDataFiles() {
assertThrows(IllegalStateException.class, writerManager::getSerializableDataFiles);
}
+ /** DataFile doesn't implement a .equals() method. Check equality manually. */
+ private static void checkDataFileEquality(DataFile d1, DataFile d2) {
Review Comment:
nit: `assertDataFileEquals` perhaps?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Managed Iceberg] Allow updating partition specs at runtime [beam]
Posted by "DanielMorales9 (via GitHub)" <gi...@apache.org>.
DanielMorales9 commented on code in PR #32879:
URL: https://github.com/apache/beam/pull/32879#discussion_r1807930482
##########
sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/RecordWriterManagerTest.java:
##########
@@ -269,6 +271,25 @@ public void testRequireClosingBeforeFetchingDataFiles() {
assertThrows(IllegalStateException.class, writerManager::getSerializableDataFiles);
}
+ /** DataFile doesn't implement a .equals() method. Check equality manually. */
+ private static void checkDataFileEquality(DataFile d1, DataFile d2) {
Review Comment:
nit: `assertDataFileEquals` perhaps?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Managed Iceberg] Allow updating partition specs at runtime [beam]
Posted by "DanielMorales9 (via GitHub)" <gi...@apache.org>.
DanielMorales9 commented on code in PR #32879:
URL: https://github.com/apache/beam/pull/32879#discussion_r1816377736
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AppendFilesToTables.java:
##########
@@ -97,36 +111,103 @@ private Catalog getCatalog() {
return catalog;
}
+ private boolean containsMultiplePartitionSpecs(Iterable<FileWriteResult> fileWriteResults) {
+ int id = fileWriteResults.iterator().next().getSerializableDataFile().getPartitionSpecId();
+ for (FileWriteResult result : fileWriteResults) {
+ if (id != result.getSerializableDataFile().getPartitionSpecId()) {
+ return true;
+ }
+ }
+ return false;
+ }
+
@ProcessElement
public void processElement(
@Element KV<String, Iterable<FileWriteResult>> element,
OutputReceiver<KV<String, SnapshotInfo>> out,
- BoundedWindow window) {
+ BoundedWindow window)
+ throws IOException {
String tableStringIdentifier = element.getKey();
Iterable<FileWriteResult> fileWriteResults = element.getValue();
if (!fileWriteResults.iterator().hasNext()) {
return;
}
Table table = getCatalog().loadTable(TableIdentifier.parse(element.getKey()));
+
+ // vast majority of the time, we will simply append data files.
+ // in the rare case we get a batch that contains multiple partition specs, we will group
+ // data into manifest files and append.
+ // note: either way, we must use a single commit operation for atomicity.
+ if (containsMultiplePartitionSpecs(fileWriteResults)) {
Review Comment:
I would rather choose between the two methods based on whether the table is partitioned or not
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Managed Iceberg] Allow updating partition specs at runtime [beam]
Posted by "ahmedabu98 (via GitHub)" <gi...@apache.org>.
ahmedabu98 commented on code in PR #32879:
URL: https://github.com/apache/beam/pull/32879#discussion_r1817281356
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AppendFilesToTables.java:
##########
@@ -97,36 +111,103 @@ private Catalog getCatalog() {
return catalog;
}
+ private boolean containsMultiplePartitionSpecs(Iterable<FileWriteResult> fileWriteResults) {
+ int id = fileWriteResults.iterator().next().getSerializableDataFile().getPartitionSpecId();
+ for (FileWriteResult result : fileWriteResults) {
+ if (id != result.getSerializableDataFile().getPartitionSpecId()) {
+ return true;
+ }
+ }
+ return false;
+ }
+
@ProcessElement
public void processElement(
@Element KV<String, Iterable<FileWriteResult>> element,
OutputReceiver<KV<String, SnapshotInfo>> out,
- BoundedWindow window) {
+ BoundedWindow window)
+ throws IOException {
String tableStringIdentifier = element.getKey();
Iterable<FileWriteResult> fileWriteResults = element.getValue();
if (!fileWriteResults.iterator().hasNext()) {
return;
}
Table table = getCatalog().loadTable(TableIdentifier.parse(element.getKey()));
+
+ // vast majority of the time, we will simply append data files.
+ // in the rare case we get a batch that contains multiple partition specs, we will group
+ // data into manifest files and append.
+ // note: either way, we must use a single commit operation for atomicity.
+ if (containsMultiplePartitionSpecs(fileWriteResults)) {
+ appendManifestFiles(table, fileWriteResults);
+ } else {
+ appendDataFiles(table, fileWriteResults);
+ }
+
+ Snapshot snapshot = table.currentSnapshot();
+ LOG.info("Created new snapshot for table '{}': {}", tableStringIdentifier, snapshot);
+ snapshotsCreated.inc();
+ out.outputWithTimestamp(
+ KV.of(element.getKey(), SnapshotInfo.fromSnapshot(snapshot)), window.maxTimestamp());
+ }
+
+ // This works only when all files are using the same partition spec.
+ private void appendDataFiles(Table table, Iterable<FileWriteResult> fileWriteResults) {
AppendFiles update = table.newAppend();
- long numFiles = 0;
for (FileWriteResult result : fileWriteResults) {
- DataFile dataFile = result.getDataFile(table.spec());
+ DataFile dataFile = result.getDataFile(table.specs());
update.appendFile(dataFile);
committedDataFileByteSize.update(dataFile.fileSizeInBytes());
committedDataFileRecordCount.update(dataFile.recordCount());
- numFiles++;
}
- // this commit will create a ManifestFile. we don't need to manually create one.
update.commit();
- dataFilesCommitted.inc(numFiles);
+ }
+
+ // When a user updates their table partition spec during runtime, we can end up with
+ // a batch of files where some are written with the old spec and some are written with the new
+ // spec.
+ // A table commit is limited to a single partition spec.
+ // To handle this, we create a manifest file for each partition spec, and group data files
+ // accordingly.
+ // Afterward, we append all manifests using a single commit operation.
+ private void appendManifestFiles(Table table, Iterable<FileWriteResult> fileWriteResults)
+ throws IOException {
+ String uuid = UUID.randomUUID().toString();
+ Map<Integer, PartitionSpec> specs = table.specs();
+ Map<Integer, ManifestWriter<DataFile>> manifestFileWriters = Maps.newHashMap();
+ // first add datafiles to the appropriate manifest file, according to its spec id
+ for (FileWriteResult result : fileWriteResults) {
Review Comment:
yep good call
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Managed Iceberg] Allow updating partition specs at runtime [beam]
Posted by "DanielMorales9 (via GitHub)" <gi...@apache.org>.
DanielMorales9 commented on code in PR #32879:
URL: https://github.com/apache/beam/pull/32879#discussion_r1816398086
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AppendFilesToTables.java:
##########
@@ -97,36 +111,103 @@ private Catalog getCatalog() {
return catalog;
}
+ private boolean containsMultiplePartitionSpecs(Iterable<FileWriteResult> fileWriteResults) {
+ int id = fileWriteResults.iterator().next().getSerializableDataFile().getPartitionSpecId();
+ for (FileWriteResult result : fileWriteResults) {
+ if (id != result.getSerializableDataFile().getPartitionSpecId()) {
+ return true;
+ }
+ }
+ return false;
+ }
+
@ProcessElement
public void processElement(
@Element KV<String, Iterable<FileWriteResult>> element,
OutputReceiver<KV<String, SnapshotInfo>> out,
- BoundedWindow window) {
+ BoundedWindow window)
+ throws IOException {
String tableStringIdentifier = element.getKey();
Iterable<FileWriteResult> fileWriteResults = element.getValue();
if (!fileWriteResults.iterator().hasNext()) {
return;
}
Table table = getCatalog().loadTable(TableIdentifier.parse(element.getKey()));
+
+ // vast majority of the time, we will simply append data files.
+ // in the rare case we get a batch that contains multiple partition specs, we will group
+ // data into manifest files and append.
+ // note: either way, we must use a single commit operation for atomicity.
+ if (containsMultiplePartitionSpecs(fileWriteResults)) {
+ appendManifestFiles(table, fileWriteResults);
+ } else {
+ appendDataFiles(table, fileWriteResults);
+ }
+
+ Snapshot snapshot = table.currentSnapshot();
+ LOG.info("Created new snapshot for table '{}': {}", tableStringIdentifier, snapshot);
+ snapshotsCreated.inc();
+ out.outputWithTimestamp(
+ KV.of(element.getKey(), SnapshotInfo.fromSnapshot(snapshot)), window.maxTimestamp());
+ }
+
+ // This works only when all files are using the same partition spec.
+ private void appendDataFiles(Table table, Iterable<FileWriteResult> fileWriteResults) {
AppendFiles update = table.newAppend();
- long numFiles = 0;
for (FileWriteResult result : fileWriteResults) {
- DataFile dataFile = result.getDataFile(table.spec());
+ DataFile dataFile = result.getDataFile(table.specs());
update.appendFile(dataFile);
committedDataFileByteSize.update(dataFile.fileSizeInBytes());
committedDataFileRecordCount.update(dataFile.recordCount());
- numFiles++;
}
- // this commit will create a ManifestFile. we don't need to manually create one.
update.commit();
- dataFilesCommitted.inc(numFiles);
+ }
+
+ // When a user updates their table partition spec during runtime, we can end up with
+ // a batch of files where some are written with the old spec and some are written with the new
+ // spec.
+ // A table commit is limited to a single partition spec.
+ // To handle this, we create a manifest file for each partition spec, and group data files
+ // accordingly.
+ // Afterward, we append all manifests using a single commit operation.
+ private void appendManifestFiles(Table table, Iterable<FileWriteResult> fileWriteResults)
Review Comment:
this is a very long method 😲
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Managed Iceberg] Allow updating partition specs at runtime [beam]
Posted by "ahmedabu98 (via GitHub)" <gi...@apache.org>.
ahmedabu98 commented on PR #32879:
URL: https://github.com/apache/beam/pull/32879#issuecomment-2456951927
@DanielMorales9 any more thoughts on this one?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Managed Iceberg] Allow updating partition specs at runtime [beam]
Posted by "DanielMorales9 (via GitHub)" <gi...@apache.org>.
DanielMorales9 commented on PR #32879:
URL: https://github.com/apache/beam/pull/32879#issuecomment-2426452525
Yep, makes sense! 👍
I checked once again and I can confirm that the manifest files are stored in the manifest list along with their Partition Spec Id and the writing process groups Data Files by their Partition Spec Id. We can merge ✅
Example Manifest list:
```{
"manifest_path": "gs://<bucket>/<schema>/<table>/metadata/c4aa6dc6-5e0b-454c-b126-1d97d2f223f8-m1.avro",
"manifest_length": 8363467,
"partition_spec_id": 0,
"content": 0,
"sequence_number": 4814,
"min_sequence_number": 4093,
"added_snapshot_id": 481579842725173797,
"added_data_files_count": 0,
"existing_data_files_count": 24356,
"deleted_data_files_count": 0,
"added_rows_count": 0,
"existing_rows_count": 785677,
"deleted_rows_count": 0,
"partitions": {
"array": []
}
}
{
"manifest_path": "gs://<bucket>/<schema>/<table>//metadata/9753c692-b48e-49bb-9c27-713c8a41842e-m1.avro",
"manifest_length": 8372685,
"partition_spec_id": 0,
"content": 0,
"sequence_number": 4093,
"min_sequence_number": 3557,
"added_snapshot_id": 1992827298894205835,
"added_data_files_count": 0,
"existing_data_files_count": 19878,
"deleted_data_files_count": 0,
"added_rows_count": 0,
"existing_rows_count": 1263244,
"deleted_rows_count": 0,
"partitions": {
"array": []
}
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Managed Iceberg] Allow updating partition specs at runtime [beam]
Posted by "ahmedabu98 (via GitHub)" <gi...@apache.org>.
ahmedabu98 commented on code in PR #32879:
URL: https://github.com/apache/beam/pull/32879#discussion_r1817291011
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AppendFilesToTables.java:
##########
@@ -97,36 +111,103 @@ private Catalog getCatalog() {
return catalog;
}
+ private boolean containsMultiplePartitionSpecs(Iterable<FileWriteResult> fileWriteResults) {
+ int id = fileWriteResults.iterator().next().getSerializableDataFile().getPartitionSpecId();
+ for (FileWriteResult result : fileWriteResults) {
+ if (id != result.getSerializableDataFile().getPartitionSpecId()) {
+ return true;
+ }
+ }
+ return false;
+ }
+
@ProcessElement
public void processElement(
@Element KV<String, Iterable<FileWriteResult>> element,
OutputReceiver<KV<String, SnapshotInfo>> out,
- BoundedWindow window) {
+ BoundedWindow window)
+ throws IOException {
String tableStringIdentifier = element.getKey();
Iterable<FileWriteResult> fileWriteResults = element.getValue();
if (!fileWriteResults.iterator().hasNext()) {
return;
}
Table table = getCatalog().loadTable(TableIdentifier.parse(element.getKey()));
+
+ // vast majority of the time, we will simply append data files.
+ // in the rare case we get a batch that contains multiple partition specs, we will group
+ // data into manifest files and append.
+ // note: either way, we must use a single commit operation for atomicity.
+ if (containsMultiplePartitionSpecs(fileWriteResults)) {
+ appendManifestFiles(table, fileWriteResults);
+ } else {
+ appendDataFiles(table, fileWriteResults);
Review Comment:
any particular reason? i'm leaning towards keeping it this way for simplicity
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Managed Iceberg] Allow updating partition specs at runtime [beam]
Posted by "DanielMorales9 (via GitHub)" <gi...@apache.org>.
DanielMorales9 commented on code in PR #32879:
URL: https://github.com/apache/beam/pull/32879#discussion_r1816386541
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AppendFilesToTables.java:
##########
@@ -97,36 +111,103 @@ private Catalog getCatalog() {
return catalog;
}
+ private boolean containsMultiplePartitionSpecs(Iterable<FileWriteResult> fileWriteResults) {
+ int id = fileWriteResults.iterator().next().getSerializableDataFile().getPartitionSpecId();
+ for (FileWriteResult result : fileWriteResults) {
+ if (id != result.getSerializableDataFile().getPartitionSpecId()) {
+ return true;
+ }
+ }
+ return false;
+ }
+
@ProcessElement
public void processElement(
@Element KV<String, Iterable<FileWriteResult>> element,
OutputReceiver<KV<String, SnapshotInfo>> out,
- BoundedWindow window) {
+ BoundedWindow window)
+ throws IOException {
String tableStringIdentifier = element.getKey();
Iterable<FileWriteResult> fileWriteResults = element.getValue();
if (!fileWriteResults.iterator().hasNext()) {
return;
}
Table table = getCatalog().loadTable(TableIdentifier.parse(element.getKey()));
+
+ // vast majority of the time, we will simply append data files.
+ // in the rare case we get a batch that contains multiple partition specs, we will group
+ // data into manifest files and append.
+ // note: either way, we must use a single commit operation for atomicity.
+ if (containsMultiplePartitionSpecs(fileWriteResults)) {
Review Comment:
See [Spark-Iceberg implmentation](https://github.com/apache/iceberg/blob/fdc2c223efa3367f8a10d9c49985326352b687b4/spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/SparkTableUtil.java#L569)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Managed Iceberg] Allow updating partition specs at runtime [beam]
Posted by "DanielMorales9 (via GitHub)" <gi...@apache.org>.
DanielMorales9 commented on code in PR #32879:
URL: https://github.com/apache/beam/pull/32879#discussion_r1817722094
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AppendFilesToTables.java:
##########
@@ -97,36 +111,103 @@ private Catalog getCatalog() {
return catalog;
}
+ private boolean containsMultiplePartitionSpecs(Iterable<FileWriteResult> fileWriteResults) {
+ int id = fileWriteResults.iterator().next().getSerializableDataFile().getPartitionSpecId();
+ for (FileWriteResult result : fileWriteResults) {
+ if (id != result.getSerializableDataFile().getPartitionSpecId()) {
+ return true;
+ }
+ }
+ return false;
+ }
+
@ProcessElement
public void processElement(
@Element KV<String, Iterable<FileWriteResult>> element,
OutputReceiver<KV<String, SnapshotInfo>> out,
- BoundedWindow window) {
+ BoundedWindow window)
+ throws IOException {
String tableStringIdentifier = element.getKey();
Iterable<FileWriteResult> fileWriteResults = element.getValue();
if (!fileWriteResults.iterator().hasNext()) {
return;
}
Table table = getCatalog().loadTable(TableIdentifier.parse(element.getKey()));
+
+ // vast majority of the time, we will simply append data files.
+ // in the rare case we get a batch that contains multiple partition specs, we will group
+ // data into manifest files and append.
+ // note: either way, we must use a single commit operation for atomicity.
+ if (containsMultiplePartitionSpecs(fileWriteResults)) {
+ appendManifestFiles(table, fileWriteResults);
+ } else {
+ appendDataFiles(table, fileWriteResults);
Review Comment:
but it's not a big duplication issue, so you can discard
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Managed Iceberg] Allow updating partition specs at runtime [beam]
Posted by "ahmedabu98 (via GitHub)" <gi...@apache.org>.
ahmedabu98 commented on code in PR #32879:
URL: https://github.com/apache/beam/pull/32879#discussion_r1817289949
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AppendFilesToTables.java:
##########
@@ -97,36 +111,103 @@ private Catalog getCatalog() {
return catalog;
}
+ private boolean containsMultiplePartitionSpecs(Iterable<FileWriteResult> fileWriteResults) {
+ int id = fileWriteResults.iterator().next().getSerializableDataFile().getPartitionSpecId();
+ for (FileWriteResult result : fileWriteResults) {
+ if (id != result.getSerializableDataFile().getPartitionSpecId()) {
+ return true;
+ }
+ }
+ return false;
+ }
+
@ProcessElement
public void processElement(
@Element KV<String, Iterable<FileWriteResult>> element,
OutputReceiver<KV<String, SnapshotInfo>> out,
- BoundedWindow window) {
+ BoundedWindow window)
+ throws IOException {
String tableStringIdentifier = element.getKey();
Iterable<FileWriteResult> fileWriteResults = element.getValue();
if (!fileWriteResults.iterator().hasNext()) {
return;
}
Table table = getCatalog().loadTable(TableIdentifier.parse(element.getKey()));
+
+ // vast majority of the time, we will simply append data files.
+ // in the rare case we get a batch that contains multiple partition specs, we will group
+ // data into manifest files and append.
+ // note: either way, we must use a single commit operation for atomicity.
+ if (containsMultiplePartitionSpecs(fileWriteResults)) {
Review Comment:
Iceberg allows us to use one commit operation that appends data files of different partitions. I think having a manifest file per partition might be a little overkill and unnecessary.
The only reason we need manifest files is to differentiate between two specs. Technically, this code path addresses an edge case, and realistically will only be used during the few moments after a table's spec gets updated -- in these moments, we will potentially have two specs in the same batch of files. So we'd need one manifest file for each spec.
I'm open to other perspectives though in case I'm missing something
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Managed Iceberg] Allow updating partition specs at runtime [beam]
Posted by "DanielMorales9 (via GitHub)" <gi...@apache.org>.
DanielMorales9 commented on code in PR #32879:
URL: https://github.com/apache/beam/pull/32879#discussion_r1816379196
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AppendFilesToTables.java:
##########
@@ -97,36 +111,103 @@ private Catalog getCatalog() {
return catalog;
}
+ private boolean containsMultiplePartitionSpecs(Iterable<FileWriteResult> fileWriteResults) {
+ int id = fileWriteResults.iterator().next().getSerializableDataFile().getPartitionSpecId();
+ for (FileWriteResult result : fileWriteResults) {
+ if (id != result.getSerializableDataFile().getPartitionSpecId()) {
+ return true;
+ }
+ }
+ return false;
+ }
+
@ProcessElement
public void processElement(
@Element KV<String, Iterable<FileWriteResult>> element,
OutputReceiver<KV<String, SnapshotInfo>> out,
- BoundedWindow window) {
+ BoundedWindow window)
+ throws IOException {
String tableStringIdentifier = element.getKey();
Iterable<FileWriteResult> fileWriteResults = element.getValue();
if (!fileWriteResults.iterator().hasNext()) {
return;
}
Table table = getCatalog().loadTable(TableIdentifier.parse(element.getKey()));
+
+ // vast majority of the time, we will simply append data files.
+ // in the rare case we get a batch that contains multiple partition specs, we will group
+ // data into manifest files and append.
+ // note: either way, we must use a single commit operation for atomicity.
+ if (containsMultiplePartitionSpecs(fileWriteResults)) {
+ appendManifestFiles(table, fileWriteResults);
+ } else {
+ appendDataFiles(table, fileWriteResults);
Review Comment:
Can we return the AppendFiles object and commit?
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AppendFilesToTables.java:
##########
@@ -97,36 +111,103 @@ private Catalog getCatalog() {
return catalog;
}
+ private boolean containsMultiplePartitionSpecs(Iterable<FileWriteResult> fileWriteResults) {
+ int id = fileWriteResults.iterator().next().getSerializableDataFile().getPartitionSpecId();
+ for (FileWriteResult result : fileWriteResults) {
+ if (id != result.getSerializableDataFile().getPartitionSpecId()) {
+ return true;
+ }
+ }
+ return false;
+ }
+
@ProcessElement
public void processElement(
@Element KV<String, Iterable<FileWriteResult>> element,
OutputReceiver<KV<String, SnapshotInfo>> out,
- BoundedWindow window) {
+ BoundedWindow window)
+ throws IOException {
String tableStringIdentifier = element.getKey();
Iterable<FileWriteResult> fileWriteResults = element.getValue();
if (!fileWriteResults.iterator().hasNext()) {
return;
}
Table table = getCatalog().loadTable(TableIdentifier.parse(element.getKey()));
+
+ // vast majority of the time, we will simply append data files.
+ // in the rare case we get a batch that contains multiple partition specs, we will group
+ // data into manifest files and append.
+ // note: either way, we must use a single commit operation for atomicity.
+ if (containsMultiplePartitionSpecs(fileWriteResults)) {
+ appendManifestFiles(table, fileWriteResults);
+ } else {
+ appendDataFiles(table, fileWriteResults);
Review Comment:
Can we return the AppendFiles update and commit?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Managed Iceberg] Allow updating partition specs at runtime [beam]
Posted by "DanielMorales9 (via GitHub)" <gi...@apache.org>.
DanielMorales9 commented on code in PR #32879:
URL: https://github.com/apache/beam/pull/32879#discussion_r1817721752
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AppendFilesToTables.java:
##########
@@ -97,36 +111,103 @@ private Catalog getCatalog() {
return catalog;
}
+ private boolean containsMultiplePartitionSpecs(Iterable<FileWriteResult> fileWriteResults) {
+ int id = fileWriteResults.iterator().next().getSerializableDataFile().getPartitionSpecId();
+ for (FileWriteResult result : fileWriteResults) {
+ if (id != result.getSerializableDataFile().getPartitionSpecId()) {
+ return true;
+ }
+ }
+ return false;
+ }
+
@ProcessElement
public void processElement(
@Element KV<String, Iterable<FileWriteResult>> element,
OutputReceiver<KV<String, SnapshotInfo>> out,
- BoundedWindow window) {
+ BoundedWindow window)
+ throws IOException {
String tableStringIdentifier = element.getKey();
Iterable<FileWriteResult> fileWriteResults = element.getValue();
if (!fileWriteResults.iterator().hasNext()) {
return;
}
Table table = getCatalog().loadTable(TableIdentifier.parse(element.getKey()));
+
+ // vast majority of the time, we will simply append data files.
+ // in the rare case we get a batch that contains multiple partition specs, we will group
+ // data into manifest files and append.
+ // note: either way, we must use a single commit operation for atomicity.
+ if (containsMultiplePartitionSpecs(fileWriteResults)) {
+ appendManifestFiles(table, fileWriteResults);
+ } else {
+ appendDataFiles(table, fileWriteResults);
Review Comment:
yep reduce duplicated code
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Managed Iceberg] Allow updating partition specs during pipeline runtime [beam]
Posted by "ahmedabu98 (via GitHub)" <gi...@apache.org>.
ahmedabu98 merged PR #32879:
URL: https://github.com/apache/beam/pull/32879
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Managed Iceberg] Allow updating partition specs at runtime [beam]
Posted by "ahmedabu98 (via GitHub)" <gi...@apache.org>.
ahmedabu98 commented on PR #32879:
URL: https://github.com/apache/beam/pull/32879#issuecomment-2426084733
> Interestingly, it seems possible to commit data files produced with an old partitioning scheme
Yep this is what I found empirically as well. The [docs](https://iceberg.apache.org/docs/1.6.1/evolution/#partition-evolution) mention that each partition spec's metadata is kept separate from each other. This allows Iceberg to carry out a separate query plan for each partition spec, then combine all files afterwards.
Old data written with the old partition spec doesn't get affected. What I'm seeing is _new_ data written with the old spec is also not affected. This is convenient for streaming writes because after updating the table's partition spec, there may be some inflight data that is still using the old partition spec. Iceberg will still accept that data when it gets committed and manage the metadata accordingly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Managed Iceberg] Allow updating partition specs at runtime [beam]
Posted by "DanielMorales9 (via GitHub)" <gi...@apache.org>.
DanielMorales9 commented on code in PR #32879:
URL: https://github.com/apache/beam/pull/32879#discussion_r1816403347
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AppendFilesToTables.java:
##########
@@ -97,36 +111,103 @@ private Catalog getCatalog() {
return catalog;
}
+ private boolean containsMultiplePartitionSpecs(Iterable<FileWriteResult> fileWriteResults) {
+ int id = fileWriteResults.iterator().next().getSerializableDataFile().getPartitionSpecId();
+ for (FileWriteResult result : fileWriteResults) {
+ if (id != result.getSerializableDataFile().getPartitionSpecId()) {
+ return true;
+ }
+ }
+ return false;
+ }
+
@ProcessElement
public void processElement(
@Element KV<String, Iterable<FileWriteResult>> element,
OutputReceiver<KV<String, SnapshotInfo>> out,
- BoundedWindow window) {
+ BoundedWindow window)
+ throws IOException {
String tableStringIdentifier = element.getKey();
Iterable<FileWriteResult> fileWriteResults = element.getValue();
if (!fileWriteResults.iterator().hasNext()) {
return;
}
Table table = getCatalog().loadTable(TableIdentifier.parse(element.getKey()));
+
+ // vast majority of the time, we will simply append data files.
+ // in the rare case we get a batch that contains multiple partition specs, we will group
+ // data into manifest files and append.
+ // note: either way, we must use a single commit operation for atomicity.
+ if (containsMultiplePartitionSpecs(fileWriteResults)) {
+ appendManifestFiles(table, fileWriteResults);
+ } else {
+ appendDataFiles(table, fileWriteResults);
+ }
+
+ Snapshot snapshot = table.currentSnapshot();
+ LOG.info("Created new snapshot for table '{}': {}", tableStringIdentifier, snapshot);
+ snapshotsCreated.inc();
+ out.outputWithTimestamp(
+ KV.of(element.getKey(), SnapshotInfo.fromSnapshot(snapshot)), window.maxTimestamp());
+ }
+
+ // This works only when all files are using the same partition spec.
+ private void appendDataFiles(Table table, Iterable<FileWriteResult> fileWriteResults) {
AppendFiles update = table.newAppend();
- long numFiles = 0;
for (FileWriteResult result : fileWriteResults) {
- DataFile dataFile = result.getDataFile(table.spec());
+ DataFile dataFile = result.getDataFile(table.specs());
update.appendFile(dataFile);
committedDataFileByteSize.update(dataFile.fileSizeInBytes());
committedDataFileRecordCount.update(dataFile.recordCount());
- numFiles++;
}
- // this commit will create a ManifestFile. we don't need to manually create one.
update.commit();
- dataFilesCommitted.inc(numFiles);
+ }
+
+ // When a user updates their table partition spec during runtime, we can end up with
+ // a batch of files where some are written with the old spec and some are written with the new
+ // spec.
+ // A table commit is limited to a single partition spec.
+ // To handle this, we create a manifest file for each partition spec, and group data files
+ // accordingly.
+ // Afterward, we append all manifests using a single commit operation.
+ private void appendManifestFiles(Table table, Iterable<FileWriteResult> fileWriteResults)
+ throws IOException {
+ String uuid = UUID.randomUUID().toString();
+ Map<Integer, PartitionSpec> specs = table.specs();
+ Map<Integer, ManifestWriter<DataFile>> manifestFileWriters = Maps.newHashMap();
+ // first add datafiles to the appropriate manifest file, according to its spec id
+ for (FileWriteResult result : fileWriteResults) {
Review Comment:
I would have rather collected all datafiles by specId and updated the manifest in one go
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Managed Iceberg] Allow updating partition specs at runtime [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #32879:
URL: https://github.com/apache/beam/pull/32879#issuecomment-2423366327
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment `assign set of reviewers`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Managed Iceberg] Allow updating partition specs at runtime [beam]
Posted by "DanielMorales9 (via GitHub)" <gi...@apache.org>.
DanielMorales9 commented on PR #32879:
URL: https://github.com/apache/beam/pull/32879#issuecomment-2425309158
I've looked into the Iceberg codebase to check for any violations of snapshot consistency. Interestingly, it seems possible to commit data files produced with an old partitioning scheme (and therefore from a previous snapshot). Could this actually be a feature of partition evolution?! 🤔
Any thoughts?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Managed Iceberg] Allow updating partition specs at runtime [beam]
Posted by "DanielMorales9 (via GitHub)" <gi...@apache.org>.
DanielMorales9 commented on code in PR #32879:
URL: https://github.com/apache/beam/pull/32879#discussion_r1817722325
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AppendFilesToTables.java:
##########
@@ -97,36 +111,103 @@ private Catalog getCatalog() {
return catalog;
}
+ private boolean containsMultiplePartitionSpecs(Iterable<FileWriteResult> fileWriteResults) {
+ int id = fileWriteResults.iterator().next().getSerializableDataFile().getPartitionSpecId();
+ for (FileWriteResult result : fileWriteResults) {
+ if (id != result.getSerializableDataFile().getPartitionSpecId()) {
+ return true;
+ }
+ }
+ return false;
+ }
+
@ProcessElement
public void processElement(
@Element KV<String, Iterable<FileWriteResult>> element,
OutputReceiver<KV<String, SnapshotInfo>> out,
- BoundedWindow window) {
+ BoundedWindow window)
+ throws IOException {
String tableStringIdentifier = element.getKey();
Iterable<FileWriteResult> fileWriteResults = element.getValue();
if (!fileWriteResults.iterator().hasNext()) {
return;
}
Table table = getCatalog().loadTable(TableIdentifier.parse(element.getKey()));
+
+ // vast majority of the time, we will simply append data files.
+ // in the rare case we get a batch that contains multiple partition specs, we will group
+ // data into manifest files and append.
+ // note: either way, we must use a single commit operation for atomicity.
+ if (containsMultiplePartitionSpecs(fileWriteResults)) {
Review Comment:
Yes, you are right. I got confused with something else. Let's keep as it is
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AppendFilesToTables.java:
##########
@@ -97,36 +111,103 @@ private Catalog getCatalog() {
return catalog;
}
+ private boolean containsMultiplePartitionSpecs(Iterable<FileWriteResult> fileWriteResults) {
+ int id = fileWriteResults.iterator().next().getSerializableDataFile().getPartitionSpecId();
+ for (FileWriteResult result : fileWriteResults) {
+ if (id != result.getSerializableDataFile().getPartitionSpecId()) {
+ return true;
+ }
+ }
+ return false;
+ }
+
@ProcessElement
public void processElement(
@Element KV<String, Iterable<FileWriteResult>> element,
OutputReceiver<KV<String, SnapshotInfo>> out,
- BoundedWindow window) {
+ BoundedWindow window)
+ throws IOException {
String tableStringIdentifier = element.getKey();
Iterable<FileWriteResult> fileWriteResults = element.getValue();
if (!fileWriteResults.iterator().hasNext()) {
return;
}
Table table = getCatalog().loadTable(TableIdentifier.parse(element.getKey()));
+
+ // vast majority of the time, we will simply append data files.
+ // in the rare case we get a batch that contains multiple partition specs, we will group
+ // data into manifest files and append.
+ // note: either way, we must use a single commit operation for atomicity.
+ if (containsMultiplePartitionSpecs(fileWriteResults)) {
Review Comment:
Yes, you are right. I got confused with something else.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Managed Iceberg] Allow updating partition specs at runtime [beam]
Posted by "DanielMorales9 (via GitHub)" <gi...@apache.org>.
DanielMorales9 commented on PR #32879:
URL: https://github.com/apache/beam/pull/32879#issuecomment-2457066321
Nope, LGTM 👍
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Managed Iceberg] Allow updating partition specs at runtime [beam]
Posted by "ahmedabu98 (via GitHub)" <gi...@apache.org>.
ahmedabu98 commented on PR #32879:
URL: https://github.com/apache/beam/pull/32879#issuecomment-2423365333
R: @DanielMorales9
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Managed Iceberg] Allow updating partition specs at runtime [beam]
Posted by "ahmedabu98 (via GitHub)" <gi...@apache.org>.
ahmedabu98 commented on PR #32879:
URL: https://github.com/apache/beam/pull/32879#issuecomment-2427225036
@DanielMorales9 sorry but I did some experimenting and found it needed some more work, namely what was missing:
- Refreshing our cached tables, so that we can use the updated partition spec when writing
- Handling batches that contain files using old and new spec. Performing a single commit on all data files is no longer an option because commits are limited to one partition spec. However, performing multiple commits is not safe (if a bundle fails between commits, we may end up with data duplication). Instead, we can create a manifest file for each spec and append them all using one commit
I fixed this up in the previous commit baba789323e351b6dab956db9cbb49c01ee142e6, can you take another look?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Managed Iceberg] Allow updating partition specs at runtime [beam]
Posted by "DanielMorales9 (via GitHub)" <gi...@apache.org>.
DanielMorales9 commented on code in PR #32879:
URL: https://github.com/apache/beam/pull/32879#discussion_r1816377736
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AppendFilesToTables.java:
##########
@@ -97,36 +111,103 @@ private Catalog getCatalog() {
return catalog;
}
+ private boolean containsMultiplePartitionSpecs(Iterable<FileWriteResult> fileWriteResults) {
+ int id = fileWriteResults.iterator().next().getSerializableDataFile().getPartitionSpecId();
+ for (FileWriteResult result : fileWriteResults) {
+ if (id != result.getSerializableDataFile().getPartitionSpecId()) {
+ return true;
+ }
+ }
+ return false;
+ }
+
@ProcessElement
public void processElement(
@Element KV<String, Iterable<FileWriteResult>> element,
OutputReceiver<KV<String, SnapshotInfo>> out,
- BoundedWindow window) {
+ BoundedWindow window)
+ throws IOException {
String tableStringIdentifier = element.getKey();
Iterable<FileWriteResult> fileWriteResults = element.getValue();
if (!fileWriteResults.iterator().hasNext()) {
return;
}
Table table = getCatalog().loadTable(TableIdentifier.parse(element.getKey()));
+
+ // vast majority of the time, we will simply append data files.
+ // in the rare case we get a batch that contains multiple partition specs, we will group
+ // data into manifest files and append.
+ // note: either way, we must use a single commit operation for atomicity.
+ if (containsMultiplePartitionSpecs(fileWriteResults)) {
Review Comment:
I would rather distinguish between partitioned and unpartitioned table
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Managed Iceberg] Allow updating partition specs at runtime [beam]
Posted by "DanielMorales9 (via GitHub)" <gi...@apache.org>.
DanielMorales9 commented on code in PR #32879:
URL: https://github.com/apache/beam/pull/32879#discussion_r1816398086
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AppendFilesToTables.java:
##########
@@ -97,36 +111,103 @@ private Catalog getCatalog() {
return catalog;
}
+ private boolean containsMultiplePartitionSpecs(Iterable<FileWriteResult> fileWriteResults) {
+ int id = fileWriteResults.iterator().next().getSerializableDataFile().getPartitionSpecId();
+ for (FileWriteResult result : fileWriteResults) {
+ if (id != result.getSerializableDataFile().getPartitionSpecId()) {
+ return true;
+ }
+ }
+ return false;
+ }
+
@ProcessElement
public void processElement(
@Element KV<String, Iterable<FileWriteResult>> element,
OutputReceiver<KV<String, SnapshotInfo>> out,
- BoundedWindow window) {
+ BoundedWindow window)
+ throws IOException {
String tableStringIdentifier = element.getKey();
Iterable<FileWriteResult> fileWriteResults = element.getValue();
if (!fileWriteResults.iterator().hasNext()) {
return;
}
Table table = getCatalog().loadTable(TableIdentifier.parse(element.getKey()));
+
+ // vast majority of the time, we will simply append data files.
+ // in the rare case we get a batch that contains multiple partition specs, we will group
+ // data into manifest files and append.
+ // note: either way, we must use a single commit operation for atomicity.
+ if (containsMultiplePartitionSpecs(fileWriteResults)) {
+ appendManifestFiles(table, fileWriteResults);
+ } else {
+ appendDataFiles(table, fileWriteResults);
+ }
+
+ Snapshot snapshot = table.currentSnapshot();
+ LOG.info("Created new snapshot for table '{}': {}", tableStringIdentifier, snapshot);
+ snapshotsCreated.inc();
+ out.outputWithTimestamp(
+ KV.of(element.getKey(), SnapshotInfo.fromSnapshot(snapshot)), window.maxTimestamp());
+ }
+
+ // This works only when all files are using the same partition spec.
+ private void appendDataFiles(Table table, Iterable<FileWriteResult> fileWriteResults) {
AppendFiles update = table.newAppend();
- long numFiles = 0;
for (FileWriteResult result : fileWriteResults) {
- DataFile dataFile = result.getDataFile(table.spec());
+ DataFile dataFile = result.getDataFile(table.specs());
update.appendFile(dataFile);
committedDataFileByteSize.update(dataFile.fileSizeInBytes());
committedDataFileRecordCount.update(dataFile.recordCount());
- numFiles++;
}
- // this commit will create a ManifestFile. we don't need to manually create one.
update.commit();
- dataFilesCommitted.inc(numFiles);
+ }
+
+ // When a user updates their table partition spec during runtime, we can end up with
+ // a batch of files where some are written with the old spec and some are written with the new
+ // spec.
+ // A table commit is limited to a single partition spec.
+ // To handle this, we create a manifest file for each partition spec, and group data files
+ // accordingly.
+ // Afterward, we append all manifests using a single commit operation.
+ private void appendManifestFiles(Table table, Iterable<FileWriteResult> fileWriteResults)
Review Comment:
This is a long method 😲
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org