feat: Added support to write iceberg tables #5989
Conversation
malhotrashivam
added
parquet
malhotrashivam
force-pushed
the
sm-ice-write
branch
from
2a60cf8
to
8c81883
Compare
malhotrashivam
changed the title
extensions/iceberg/s3/src/main/java/io/deephaven/iceberg/util/IcebergToolsS3.java
Show resolved
Hide resolved
extensions/iceberg/src/main/java/io/deephaven/iceberg/base/IcebergUtils.java
Outdated
Show resolved
Hide resolved
extensions/iceberg/src/main/java/io/deephaven/iceberg/util/IcebergCatalogAdapter.java
Show resolved
Hide resolved
extensions/iceberg/src/main/java/io/deephaven/iceberg/util/IcebergCatalogAdapter.java
Outdated
Show resolved
Hide resolved
extensions/iceberg/src/main/java/io/deephaven/iceberg/util/IcebergCatalogAdapter.java
Outdated
Show resolved
Hide resolved
extensions/iceberg/src/main/java/io/deephaven/iceberg/util/IcebergCatalogAdapter.java
Outdated
Show resolved
Hide resolved
extensions/iceberg/s3/src/test/java/io/deephaven/iceberg/util/IcebergLocalStackTest.java
Outdated
Show resolved
Hide resolved
| @@ -33,95 +36,18 @@ | |||
| */ | |||
| public abstract class ParquetInstructions implements ColumnToCodecMappings { | |||
|
|
|||
| private static volatile String defaultCompressionCodecName = CompressionCodecName.SNAPPY.toString(); | |||
There was a problem hiding this comment.
Removing unnecessary configuration parameters.
There was a problem hiding this comment.
In general this seems like an improvement, the old code was adding little value for the complexity. What about enterprise usages?
| @@ -433,6 +382,14 @@ public boolean useDictionary() { | |||
| public void useDictionary(final boolean useDictionary) { | |||
| this.useDictionary = useDictionary; | |||
| } | |||
|
|
|||
| public OptionalInt getFieldId() { | |||
There was a problem hiding this comment.
The field Id related logic may change when #6156 gets merged.
extensions/iceberg/src/main/java/io/deephaven/iceberg/util/IcebergCatalogAdapter.java
Outdated
Show resolved
Hide resolved
extensions/iceberg/s3/src/main/java/io/deephaven/iceberg/util/IcebergToolsS3.java
Show resolved
Hide resolved
extensions/iceberg/src/main/java/io/deephaven/iceberg/base/IcebergUtils.java
Outdated
Show resolved
Hide resolved
extensions/iceberg/src/main/java/io/deephaven/iceberg/util/IcebergCatalogAdapter.java
Show resolved
Hide resolved
extensions/iceberg/src/main/java/io/deephaven/iceberg/util/IcebergCatalogAdapter.java
Outdated
Show resolved
Hide resolved
| maximum_dictionary_size: Optional[int] = None, | ||
| target_page_size: Optional[int] = None, | ||
| verify_schema: Optional[bool] = None, | ||
| dh_to_iceberg_column_renames: Optional[Dict[str, str]] = None, |
There was a problem hiding this comment.
name is very long, especially if a user is specifying it. Any reason it can't just be column_renames?
There was a problem hiding this comment.
you should also look through the rest of the API to see if column_renames or col_renames would be most consistent. I would guess col_renames.
| if compression_codec_name is not None: | ||
| builder.compressionCodecName(compression_codec_name) | ||
|
|
||
| if maximum_dictionary_keys is not None: | ||
| builder.maximumDictionaryKeys(maximum_dictionary_keys) | ||
|
|
||
| if maximum_dictionary_size is not None: | ||
| builder.maximumDictionarySize(maximum_dictionary_size) | ||
|
|
||
| if target_page_size is not None: | ||
| builder.targetPageSize(target_page_size) | ||
|
|
||
| if verify_schema is not None: | ||
| builder.verifySchema(verify_schema) | ||
|
|
||
| if dh_to_iceberg_column_renames is not None: | ||
| for dh_name, iceberg_name in dh_to_iceberg_column_renames.items(): | ||
| builder.putDhToIcebergColumnRenames(dh_name, iceberg_name) | ||
|
|
||
| if table_definition is not None: | ||
| builder.tableDefinition(TableDefinition(table_definition).j_table_definition) | ||
|
|
||
| if data_instructions is not None: | ||
| builder.dataInstructions(data_instructions.j_object) |
There was a problem hiding this comment.
I suspect all of these cases can have is not None removed. Confirm with @jmao-denver on what he wants to see.
| tables: List[Table], | ||
| partition_paths: Optional[List[str]] = None, | ||
| instructions: Optional[IcebergParquetWriteInstructions] = None): | ||
| # TODO Review javadoc in this file once again |
| table_identifier: str, | ||
| tables: List[Table], | ||
| partition_paths: Optional[List[str]] = None, | ||
| instructions: Optional[IcebergParquetWriteInstructions] = None): |
| instructions: Optional[IcebergParquetWriteInstructions] = None): | ||
| # TODO Review javadoc in this file once again | ||
| """ | ||
| Append the provided Deephaven table as a new partition to the existing Iceberg table in a single snapshot. This |
There was a problem hiding this comment.
this says "table" and "partition", but the input is a list of tables. Does that mean multiple tables go to one partition or multiple partitions? etc.
| tables: List[Table], | ||
| partition_paths: Optional[List[str]] = None, |
| table_identifier: str, | ||
| tables: List[Table], | ||
| partition_paths: Optional[List[str]] = None, | ||
| instructions: Optional[IcebergParquetWriteInstructions] = None): |
| of data files that were written. Users can use this list to create a transaction/snapshot if needed. | ||
|
|
||
| Args: | ||
| table_identifier (str): the identifier string for iceberg table to write to. |
| tables (List[Table]): the tables to write. | ||
| partition_paths (Optional[List[str]]): the partitioning path at which data would be written, for example, | ||
| "year=2021/month=01". If omitted, we will try to write data to the table without partitioning. |
| partition_paths (Optional[List[str]]): the partitioning path at which data would be written, for example, | ||
| "year=2021/month=01". If omitted, we will try to write data to the table without partitioning. | ||
| instructions (Optional[IcebergParquetWriteInstructions]): the instructions for customizations while writing. | ||
| """ |
There was a problem hiding this comment.
All above cases that are missing the return type hint are also missing docs on the return value
extensions/iceberg/src/main/java/io/deephaven/iceberg/base/IcebergUtils.java
Outdated
Show resolved
Hide resolved
extensions/iceberg/src/main/java/io/deephaven/iceberg/base/IcebergUtils.java
Show resolved
Hide resolved
extensions/iceberg/src/main/java/io/deephaven/iceberg/base/IcebergUtils.java
Outdated
Show resolved
Hide resolved
extensions/iceberg/src/main/java/io/deephaven/iceberg/base/IcebergUtils.java
Outdated
Show resolved
Hide resolved
extensions/iceberg/src/main/java/io/deephaven/iceberg/base/IcebergUtils.java
Outdated
Show resolved
Hide resolved
extensions/iceberg/src/main/java/io/deephaven/iceberg/util/IcebergBaseInstructions.java
Show resolved
Hide resolved
extensions/iceberg/src/main/java/io/deephaven/iceberg/util/IcebergTableAdapter.java
Outdated
Show resolved
Hide resolved
extensions/iceberg/src/main/java/io/deephaven/iceberg/util/IcebergTableAdapter.java
Outdated
Show resolved
Hide resolved
extensions/iceberg/src/main/java/io/deephaven/iceberg/util/IcebergTableAdapter.java
Outdated
Show resolved
Hide resolved
extensions/iceberg/src/main/java/io/deephaven/iceberg/util/IcebergTableAdapter.java
Outdated
Show resolved
Hide resolved
| * The Deephaven tables to be written. All tables should have the same definition, else a {@link #tableDefinition() | ||
| * table definition} should be provided. | ||
| */ | ||
| public abstract List<Table> dhTables(); |
There was a problem hiding this comment.
Not a fan generally of prefixing based on namespaces like "dh"; it's okay sometimes as a variable, but I would just let the return type speak for itself io.deephaven.engine.table.Table and call this tables().
There was a problem hiding this comment.
Do all of these tables need to have exactly #tableDefinition if that is present? If so, we should add a check.
There was a problem hiding this comment.
All tables do not need to have exactly the same definition if #tableDefinition is provided, else they need to have same definition. I have these check in IcebergTableAdapter::ensureDefinition, I can move some of that logic to the Instructions class.
I have also updated the docs for #tableDefinition to make it more clear.
extensions/iceberg/src/main/java/io/deephaven/iceberg/util/IcebergWriteInstructions.java
Outdated
Show resolved
Hide resolved
extensions/iceberg/src/main/java/io/deephaven/iceberg/util/TableWriterOptions.java
Show resolved
Hide resolved
| * <p> | ||
| * If not provided, we use the latest schema from the table. | ||
| */ | ||
| public abstract Optional<Schema> schema(); |
There was a problem hiding this comment.
I was debating arguing for the point
We may want to enforce that schema is present if the map is present, because it's Iceberg's responsibility to set the field ids when the schema is created or updated, and thus a user should only be getting the field IDs from an existing schema
but upon further reflection, it's possible that the users are either hard-coding just field IDs after the fact (ie, after the table has been created), or getting it from some other system that interacts with Iceberg.
Of course, in that scenario, you might argue that they should be hard-coding the Schema as opposed to just the field ids... this also does up a bit of a bootstrapping chicken and an egg problem - how can a piece of writing logic be written that is both responsible for creating an initial table if it doesn't exist, but also appending it to it successfully if the table does exist? Are we allowed to assume that Iceberg will create the Schema with the field IDs incrementing starting from 1? We should discuss this bootstrapping assumption problem, and how we may best need to solve it...
I think this method should move above the map; it feels more structurally important.
There was a problem hiding this comment.
It looks like we might be able to make assumptions about the ordering. Digging into the code I see org.apache.iceberg.types.TypeUtil#assignFreshIds(int, org.apache.iceberg.Schema, org.apache.iceberg.types.TypeUtil.NextID); of course, this is only the de-facto implementation, not sure if the field ids are guaranteed to be in this order based on the spec...
There was a problem hiding this comment.
Another argument, in the case where DH is the only writer, is the user should be able to completely ignore schema and the map...
extensions/iceberg/src/main/java/io/deephaven/iceberg/util/IcebergTableAdapter.java
Outdated
Show resolved
Hide resolved
extensions/iceberg/src/test/java/io/deephaven/iceberg/junit5/SqliteCatalogBase.java
Outdated
Show resolved
Hide resolved
| // Sleep for 0.5 second | ||
| try { | ||
| Thread.sleep(500); | ||
| } catch (InterruptedException e) { | ||
| Thread.currentThread().interrupt(); | ||
| } |
There was a problem hiding this comment.
Ideally, we should have some sort of testing hooks so we don't need to do any sleeps; or maybe we already have a way to wait for new data against the table?
Regardless, if a thread is interrupted in this way in a test, probably better to just add it as exception to the test method.
There was a problem hiding this comment.
I checked with Larry, he said he couldn't find a better way to test this part, so I left it like this for now.
Will check with Ryan.
extensions/iceberg/src/test/java/io/deephaven/iceberg/junit5/SqliteCatalogBase.java
Outdated
Show resolved
Hide resolved
| /** | ||
| * @return A callback to be executed when on completing each parquet data file write (excluding the index and | ||
| * metadata files). | ||
| */ | ||
| public abstract Optional<OnWriteCompleted> onWriteCompleted(); |
There was a problem hiding this comment.
Based on the current implementation, I see this always gets invoked on-thread in a linear fashion; we may want to document that the consumer is responsible for thread-safety, or that the writing code will invoke this in a thread safe way. Both ways have there merit, not sure which I prefer. @rcaudy ?
extensions/parquet/table/src/main/java/io/deephaven/parquet/table/ParquetInstructions.java
Outdated
Show resolved
Hide resolved
extensions/iceberg/src/main/java/io/deephaven/iceberg/util/IcebergWriteInstructions.java
Show resolved
Hide resolved
| if (numTables == 0) { | ||
| return writeInstructions.withTableDefinition(TableDefinition.of()); | ||
| } | ||
| final List<Table> dhTables = writeInstructions.tables(); |
There was a problem hiding this comment.
So, I get the spirit of this method, but I'm not a fan of how we now have 2 classes of IcebergParquetWriteInstructions. It's the same sort of situation I'm sad about wrt #6149.
I wonder instead of we should add a method
public abstract class IcebergWriteInstructions implements IcebergBaseInstructions {
...
public final TableDefinition tableDefinitionOrFirst() {
return tableDefinition().orElse(tables().get(0).getDefinition());
}
...and then, where applicable, have callers use tableDefinitionOrFirst instead of tableDefinition. This saves us from having to create a new object and allows us to preserve the original instructions further down through the call stack.
extensions/iceberg/src/test/java/io/deephaven/iceberg/junit5/SqliteCatalogBase.java
Show resolved
Hide resolved
extensions/iceberg/src/test/java/io/deephaven/iceberg/junit5/SqliteCatalogBase.java
Show resolved
Hide resolved
extensions/iceberg/src/test/java/io/deephaven/iceberg/junit5/SqliteCatalogBase.java
Show resolved
Hide resolved
extensions/parquet/table/src/main/java/io/deephaven/parquet/table/CompletedParquetWrite.java
Show resolved
Hide resolved
| writeInstructions.onWriteCompleted() | ||
| .ifPresent(callback -> callback.onWriteCompleted(CompletedParquetWrite.builder() | ||
| .destination(tableDestination) | ||
| .numRows(source.size()) |
There was a problem hiding this comment.
I wonder if it's more appropriate to plumb the callback down through ParquetTableWriter.write? In that way, write doesn't need to return the number of bytes anymore, and it's going to be responsible internally for calling the number of rows (instead of making the caller do source.size(); I do see the safety check is at the inner layer calling checkInitiateSerialTableOperation).
| public IcebergTableAdapter createTable( | ||
| @NotNull final TableIdentifier tableIdentifier, | ||
| @NotNull final TableDefinition definition) { | ||
| // TODO Add these APIs to python code once finalized |
|
|
||
| INSTRUCTIONS_BUILDER addAllDhTables(Iterable<? extends Table> elements); | ||
|
|
||
| // TODO Discuss about the API for partition paths, and add tests |
There was a problem hiding this comment.
TODO Check with Devin if this is okay.
| * The Deephaven tables to be written. All tables should have the same definition, else a {@link #tableDefinition() | ||
| * table definition} should be provided. | ||
| */ | ||
| public abstract List<Table> dhTables(); |
There was a problem hiding this comment.
All tables do not need to have exactly the same definition if #tableDefinition is provided, else they need to have same definition. I have these check in IcebergTableAdapter::ensureDefinition, I can move some of that logic to the Instructions class.
I have also updated the docs for #tableDefinition to make it more clear.
| // Sleep for 0.5 second | ||
| try { | ||
| Thread.sleep(500); | ||
| } catch (InterruptedException e) { | ||
| Thread.currentThread().interrupt(); | ||
| } |
There was a problem hiding this comment.
I checked with Larry, he said he couldn't find a better way to test this part, so I left it like this for now.
Will check with Ryan.
| // Overwrite with an empty table | ||
| final Table emptyTable = TableTools.emptyTable(0) | ||
| .update("intCol = (int) 4 * i + 30", | ||
| "doubleCol = (double) 4.5 * i + 30"); | ||
| tableAdapter.overwrite(instructionsBuilder() | ||
| .addDhTables(emptyTable) | ||
| .build()); |
There was a problem hiding this comment.
Yea, it now seems like an unnecessary complication.
If user wants to delete the content, they can do it with just the iceberg API and don't need Deephaven to help.
So I can delete it for now and we can add it later if needed.
extensions/iceberg/src/main/java/io/deephaven/iceberg/util/IcebergTableWriter.java
Outdated
Show resolved
Hide resolved
| * The {@link TableDefinition} to use when writing Iceberg data files. All tables written by this writer should have | ||
| * the same definition. | ||
| */ | ||
| public abstract TableDefinition tableDefinition(); |
There was a problem hiding this comment.
So this definition can be a subset or a superset as well.
This is similar to how we treat table definition on parquet reading/writing side too, where adding an additional column here will lead to null values in the table.
I have updated the javadocs to make it more clear.
extensions/iceberg/src/main/java/io/deephaven/iceberg/util/IcebergTableWriter.java
Outdated
Show resolved
Hide resolved
Closes: #6125
Should be merged after #6156
Also moves existing Iceberg tests from Junit4 to Junit5.