Spark: Support spec ID and partition metadata columns #2984

aokolnychyi · 2021-08-16T19:03:38Z

This WIP PR adds support for _spec and _partition metadata columns so that we can project them for writing deltas.

aokolnychyi · 2021-08-16T19:25:15Z

core/src/main/java/org/apache/iceberg/MetadataColumns.java

      FILE_PATH.name(), FILE_PATH,
      ROW_POSITION.name(), ROW_POSITION,
      IS_DELETED.name(), IS_DELETED);

-  private static final Set<Integer> META_IDS = META_COLUMNS.values().stream().map(NestedField::fieldId)


Had to change this as the partition type is not static and is handled specially.

aokolnychyi · 2021-08-16T19:26:47Z

core/src/main/java/org/apache/iceberg/MetadataColumns.java

-  public static NestedField get(String name) {
-    return META_COLUMNS.get(name);
+  public static NestedField metadataColumn(Table table, String name) {
+    if (name.equals(PARTITION_COLUMN_NAME)) {


I kept the logic case sensitive as before but we maybe should reconsider it at some point.

core/src/main/java/org/apache/iceberg/Partitioning.java

aokolnychyi · 2021-08-16T19:27:38Z

core/src/main/java/org/apache/iceberg/util/PartitionUtil.java

+    // add _partition
+    if (partitionType.fields().isEmpty()) {
+      // use null as some query engines may not be able to handle empty structs
+      idToConstant.put(MetadataColumns.PARTITION_COLUMN_ID, null);


Using null for unpartitioned tables.

flink/src/main/java/org/apache/iceberg/flink/source/FlinkInputFormat.java

aokolnychyi · 2021-08-16T19:29:59Z

spark/src/main/java/org/apache/iceberg/spark/source/BaseDataReader.java

+        StructType structType = (StructType) type;
+
+        if (structType.fields().isEmpty()) {
+          return new GenericInternalRow();


We won't hit this clause with _partition as the passed value will be null.

aokolnychyi · 2021-08-16T19:30:34Z

spark3/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java

@@ -149,7 +149,7 @@ private Schema schemaWithMetadataColumns() {
    // metadata columns
    List<Types.NestedField> fields = metaColumns.stream()
        .distinct()
-        .map(MetadataColumns::get)
+        .map(name -> MetadataColumns.metadataColumn(table, name))


Need to pass a Table object to figure our the partition type.

aokolnychyi · 2021-08-16T19:31:07Z

spark3/src/test/java/org/apache/iceberg/spark/source/SparkTestTable.java

+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.sql.util.CaseInsensitiveStringMap;
+
+public class SparkTestTable extends SparkTable {


This is purely a temp solution for testing purposes until we compile against 3.2.

aokolnychyi · 2021-08-16T19:31:40Z

spark3/src/test/java/org/apache/iceberg/spark/source/TestSparkCatalog.java

@@ -31,7 +30,14 @@

  @Override
  public Table loadTable(Identifier ident) throws NoSuchTableException {
-    TestTables.TestTable table = TestTables.load(Spark3Util.identifierToTableIdentifier(ident).toString());
-    return new SparkTable(table, false);
+    String[] parts = ident.name().split("\\$", 2);


I probably need to add a comment here. This is also for testing purposes until we consume Spark 3.2.

aokolnychyi · 2021-08-16T19:32:07Z

spark3/src/test/java/org/apache/iceberg/spark/source/TestSparkMetadataColumns.java

+  @Test
+  public void testSpecAndPartitionMetadataColumns() {
+    // TODO: support metadata structs in vectorized ORC reads
+    Assume.assumeFalse(fileFormat == FileFormat.ORC && vectorized);


ORC vectorized path does not support constant structs.

flink/src/main/java/org/apache/iceberg/flink/source/FlinkInputFormat.java

aokolnychyi · 2021-08-16T20:28:45Z

cc @openinx @stevenzwu @rdblue @rdsr @pvary @RussellSpitzer @flyrain @szehon-ho @karuppayya @kbendick @jackye1995 @chenjunjiedada

core/src/main/java/org/apache/iceberg/MetadataColumns.java

core/src/main/java/org/apache/iceberg/util/PartitionUtil.java

core/src/main/java/org/apache/iceberg/Partitioning.java

core/src/main/java/org/apache/iceberg/MetadataColumns.java

mr/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergInputFormat.java

core/src/main/java/org/apache/iceberg/MetadataColumns.java

jackye1995 · 2021-08-17T05:48:17Z

core/src/main/java/org/apache/iceberg/MetadataColumns.java

  }

  public static boolean isMetadataColumn(String name) {
-    return META_COLUMNS.containsKey(name);
+    return name.equals(PARTITION_COLUMN_NAME) || META_COLUMNS.containsKey(name);


I am starting to wonder if it is better to just have a dummy column type mapping for _partition in META_COLUMNS and only separate the logic when necessary, which can avoid changes like this in quite a few places.

I did that first but META_COLUMNS is an immutable map that does not allow null keys. I am reluctant to switch to a mutable map so I added this condition here. I hope we will be able to use metadataColumn(table, name) in other places so this workaround will be only part of MetadataColumns.

It looks ugly, though, I agree.

core/src/main/java/org/apache/iceberg/Partitioning.java

aokolnychyi · 2021-09-22T23:35:47Z

I've updated this PR to focus on Spark for now. Similar functionality can be added to other query engines later.

aokolnychyi · 2021-09-22T23:36:11Z

I'd appreciate another review round.

aokolnychyi · 2021-09-23T01:13:16Z

This is no longer WIP.

api/src/main/java/org/apache/iceberg/util/StructProjection.java

rdblue · 2021-09-26T20:06:23Z

api/src/main/java/org/apache/iceberg/util/StructProjection.java

+    if (structPos != -1) {
+      return struct.get(structPos, javaClass);
+    } else {
+      return null;


Since this relies on a null for nestedProjections[pos], I think it should set nestedProjections[pos] to null in the constructor where positionMap[pos] is set to -1.

I think it is actually always null as long as the field was not found. I'll add an explicit call, though.

core/src/main/java/org/apache/iceberg/MetadataColumns.java

rdblue · 2021-09-26T20:12:14Z

core/src/main/java/org/apache/iceberg/MetadataColumns.java

+  public static final NestedField SPEC_ID = NestedField.required(
+      Integer.MAX_VALUE - 4, "_spec_id", Types.IntegerType.get(), "Spec ID to which a row belongs to");
+  // the partition column type is not static and depends on all specs in the table
+  public static final int PARTITION_COLUMN_ID = Integer.MAX_VALUE - 5;


I think that new reserved columns need to be added to the spec. We should also make sure the spec notes the ranges that are reserved. I'm not sure if we did that or just added specific IDs.

I'll do that in a follow-up as we are missing _deleted too.

rdblue · 2021-09-26T20:16:18Z

core/src/main/java/org/apache/iceberg/util/PartitionUtil.java

+    // add _partition
+    if (partitionType != null && partitionType.fields().size() > 0) {
+      StructLike coercedPartition = coercePartition(partitionType, spec, partitionData);
+      idToConstant.put(MetadataColumns.PARTITION_COLUMN_ID, convertConstant.apply(partitionType, coercedPartition));


This needs to call convertConstant on each partition value. The assumption in all of the implementations is that it is called on primitive values, not on structs.

Okay, I see what's happening. This PR updates the conversion for Spark and because most callers only use constantsMap(task, func) there's no way to get a partition passed through. I'm a little uneasy about this, but since we need to convert to a specific class for the reader I don't see a good way around it.

Yeah, it is also related to unknown transforms.

core/src/main/java/org/apache/iceberg/util/PartitionUtil.java

rdblue · 2021-09-26T20:20:50Z

spark/src/main/java/org/apache/iceberg/spark/source/BaseDataReader.java

+  protected Map<Integer, ?> constantsMap(FileScanTask task, Schema readSchema) {
+    if (readSchema.findField(MetadataColumns.PARTITION_COLUMN_ID) != null) {
+      StructType partitionType = Partitioning.partitionType(table);
+      return PartitionUtil.constantsMap(task, partitionType, BaseDataReader::convertConstant);


I'd prefer just to call the method rather than trying to optimize by not adding the partition entry.

I'd prefer calling this all the time too. However, this would mean we can no longer query tables with unknown transforms. We need to know all transforms to build a common partition type. That's why I don't set the partition column if it is not requested.

Any thoughts on this, @rdblue?

Can we leave unknown transforms out of the partition type instead? We can just ignore them if they're unknown?

I'd probably consider persisting the partition type in the metadata instead. It might be confusing to silently ignore a partition column.

I think it would be fine for unknown partitions right now. It would unblock this without much risk.

rdblue · 2021-09-26T20:26:46Z

spark3/src/test/java/org/apache/iceberg/spark/source/TestSparkCatalog.java

+    if (parts.length == 2) {
+      TestTables.TestTable table = TestTables.load(parts[0]);
+      String[] metadataColumns = parts[1].split(",");
+      return new SparkTestTable(table, metadataColumns, false);


This is just a different way to pass metadata columns in? Is there a better way to test it?

I wanted to simply overload newScanBuilder in SparkTestTable but it did not work. We can get rid of this once we migrate to Spark 3.2 so I don't worry too much.

rdblue · 2021-09-26T20:29:26Z

spark3/src/test/java/org/apache/iceberg/spark/source/TestSparkMetadataColumns.java

+        row(3, row(null, 2))
+    );
+    assertEquals("Rows must match", expected,
+        sql("SELECT _spec_id, _partition FROM `%s$_spec_id,_partition` ORDER BY _spec_id", TABLE_NAME));


This should probably have a TODO comment to fix it when we can rely on Spark 3.2.

Added TODOs before all tests and in SparkTestTable.

spark/src/main/java/org/apache/iceberg/spark/source/BaseDataReader.java

rdblue

I noted a few minor things, but overall this is good. Please fix some of those things and then feel free to merge.

aokolnychyi · 2021-09-27T18:42:00Z

Thanks for reviewing, @rdblue @jackye1995 @RussellSpitzer @kbendick @stevenzwu!

github-actions bot added core data flink MR spark labels Aug 16, 2021

aokolnychyi commented Aug 16, 2021

View reviewed changes

core/src/main/java/org/apache/iceberg/Partitioning.java Outdated Show resolved Hide resolved

aokolnychyi commented Aug 16, 2021

View reviewed changes

flink/src/main/java/org/apache/iceberg/flink/source/FlinkInputFormat.java Outdated Show resolved Hide resolved

aokolnychyi commented Aug 16, 2021

View reviewed changes

flink/src/main/java/org/apache/iceberg/flink/source/FlinkInputFormat.java Outdated Show resolved Hide resolved

rdblue reviewed Aug 16, 2021

View reviewed changes

core/src/main/java/org/apache/iceberg/MetadataColumns.java Outdated Show resolved Hide resolved

rdblue reviewed Aug 16, 2021

View reviewed changes

core/src/main/java/org/apache/iceberg/util/PartitionUtil.java Outdated Show resolved Hide resolved

RussellSpitzer reviewed Aug 16, 2021

View reviewed changes

core/src/main/java/org/apache/iceberg/Partitioning.java Outdated Show resolved Hide resolved

kbendick reviewed Aug 16, 2021

View reviewed changes

core/src/main/java/org/apache/iceberg/MetadataColumns.java Show resolved Hide resolved

mr/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergInputFormat.java Outdated Show resolved Hide resolved

stevenzwu reviewed Aug 17, 2021

View reviewed changes

core/src/main/java/org/apache/iceberg/MetadataColumns.java Outdated Show resolved Hide resolved

jackye1995 reviewed Aug 17, 2021

View reviewed changes

core/src/main/java/org/apache/iceberg/MetadataColumns.java Outdated Show resolved Hide resolved

jackye1995 reviewed Aug 17, 2021

View reviewed changes

core/src/main/java/org/apache/iceberg/Partitioning.java Outdated Show resolved Hide resolved

This was referenced Aug 17, 2021

Flink: Use SerializableTable in FlinkInputFormat #2987

Closed

MR: Use SerializableTable in IcebergSplit #2988

Merged

Spark: Support spec ID and partition metadata columns

480c1be

aokolnychyi force-pushed the spec-partition-metadata-cols-v2 branch from 71c3c64 to 480c1be Compare September 22, 2021 23:34

github-actions bot added the API label Sep 22, 2021

aokolnychyi changed the title ~~Core: Support spec and partition metadata columns~~ Spark: Support spec ID and partition metadata columns Sep 22, 2021

aokolnychyi marked this pull request as ready for review September 23, 2021 01:13

rdblue reviewed Sep 26, 2021

View reviewed changes

api/src/main/java/org/apache/iceberg/util/StructProjection.java Outdated Show resolved Hide resolved

rdblue reviewed Sep 26, 2021

View reviewed changes

core/src/main/java/org/apache/iceberg/MetadataColumns.java Outdated Show resolved Hide resolved

rdblue reviewed Sep 26, 2021

View reviewed changes

core/src/main/java/org/apache/iceberg/util/PartitionUtil.java Outdated Show resolved Hide resolved

rdblue reviewed Sep 26, 2021

View reviewed changes

spark/src/main/java/org/apache/iceberg/spark/source/BaseDataReader.java Show resolved Hide resolved

rdblue approved these changes Sep 26, 2021

View reviewed changes

Review comments

ac34d21

aokolnychyi merged commit 7cf96f0 into apache:master Sep 27, 2021

This was referenced Sep 27, 2021

Spec: Add _deleted, _spec_id, _partition metadata columns #3190

Merged

ORC: support metadata column structs with vectorization #3192

Closed

kbendick pushed a commit to kbendick/iceberg that referenced this pull request Nov 2, 2021

Spark: Support spec ID and partition metadata columns (apache#2984)

36aebe2

kbendick mentioned this pull request Nov 2, 2021

Investigate amount of work needed to backport #3240 to 0.12.1 #3443

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Spark: Support spec ID and partition metadata columns #2984

Spark: Support spec ID and partition metadata columns #2984

aokolnychyi commented Aug 16, 2021

aokolnychyi Aug 16, 2021

aokolnychyi Aug 16, 2021

aokolnychyi Aug 16, 2021

aokolnychyi Aug 16, 2021

aokolnychyi Aug 16, 2021

aokolnychyi Aug 16, 2021

aokolnychyi Aug 16, 2021

aokolnychyi Aug 16, 2021

aokolnychyi commented Aug 16, 2021 •

edited

Loading

jackye1995 Aug 17, 2021

aokolnychyi Aug 17, 2021

aokolnychyi commented Sep 22, 2021

aokolnychyi commented Sep 22, 2021

aokolnychyi commented Sep 23, 2021

rdblue Sep 26, 2021

aokolnychyi Sep 27, 2021

aokolnychyi Sep 27, 2021

rdblue Sep 26, 2021

aokolnychyi Sep 27, 2021

rdblue Sep 26, 2021

rdblue Sep 26, 2021

aokolnychyi Sep 27, 2021 •

edited

Loading

rdblue Sep 26, 2021

aokolnychyi Sep 26, 2021

rdblue Sep 26, 2021

aokolnychyi Sep 27, 2021

rdblue Sep 27, 2021

rdblue Sep 26, 2021

aokolnychyi Sep 27, 2021

rdblue Sep 26, 2021

aokolnychyi Sep 27, 2021 •

edited

Loading

rdblue left a comment

aokolnychyi commented Sep 27, 2021

Spark: Support spec ID and partition metadata columns #2984

Spark: Support spec ID and partition metadata columns #2984

Conversation

aokolnychyi commented Aug 16, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

aokolnychyi commented Aug 16, 2021 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

aokolnychyi commented Sep 22, 2021

aokolnychyi commented Sep 22, 2021

aokolnychyi commented Sep 23, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

aokolnychyi Sep 27, 2021 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

aokolnychyi Sep 27, 2021 • edited Loading

Choose a reason for hiding this comment

rdblue left a comment

Choose a reason for hiding this comment

aokolnychyi commented Sep 27, 2021

aokolnychyi commented Aug 16, 2021 •

edited

Loading

aokolnychyi Sep 27, 2021 •

edited

Loading

aokolnychyi Sep 27, 2021 •

edited

Loading