Skip to content
New issue

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

Sign up for GitHub

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

Already on GitHub? Sign in to your account

Spark: Add some tests for variant fixup #12497

Open
wants to merge 7 commits into
base: main
Choose a base branch
from
Open

Spark: Add some tests for variant fixup #12497

Show file tree
Hide file tree
Changes from 6 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
a43bf5c
Spark: Add some tests for variant fixup
XBaith Mar 11, 2025
23f7291
Implement variant in PruneColumnsWithoutReordering and add UnknownTyp…
XBaith Mar 12, 2025
f036b59
Add variant method in Spark3.4 PruneColumnsWithoutReordering
XBaith Mar 14, 2025
74c5118
Add support for VariantType in SparkFixupTypes tests
XBaith Mar 17, 2025
c64903c
Refactor RandomData and TestSparkFixupTypes: remove unused variant me…
XBaith Mar 19, 2025
01c58d0
Remove variant in spark3.4
XBaith Mar 19, 2025
347be8b
Fixup to VARIANT type in BINARY checks
XBaith Mar 20, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/PruneColumnsWithoutReordering.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -193,6 +193,11 @@ public Type map(Types.MapType map, Supplier<Type> keyResult, Supplier<Type> valu
}
}

@Override
public Type variant(Types.VariantType variant) {
return variant;
}

@Override
public Type primitive(Type.PrimitiveType primitive) {
Set<Class<? extends DataType>> expectedType = TYPES.get(primitive.typeId());
Expand Down
134 changes: 134 additions & 0 deletions spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/TestSparkFixupTypes.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,134 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
package org.apache.iceberg.spark;

import static org.assertj.core.api.Assertions.assertThat;

import org.apache.iceberg.Schema;
import org.apache.iceberg.types.Types;
import org.junit.jupiter.api.Test;

public class TestSparkFixupTypes {

@Test
void fixupShouldCorrectlyFixUUID() {
Schema schema = new Schema(Types.NestedField.required(1, "field", Types.StringType.get()));
Schema referenceSchema =
new Schema(Types.NestedField.required(1, "field", Types.UUIDType.get()));

Schema fixedSchema = SparkFixupTypes.fixup(schema, referenceSchema);

assertThat(fixedSchema.findType("field")).isSameAs(Types.UUIDType.get());
}

@Test
void fixupShouldCorrectlyFixBinary() {
Schema schema = new Schema(Types.NestedField.required(1, "field", Types.BinaryType.get()));
Schema referenceSchema =
new Schema(Types.NestedField.required(1, "field", Types.FixedType.ofLength(16)));

Schema fixedSchema = SparkFixupTypes.fixup(schema, referenceSchema);

assertThat(fixedSchema.findType("field")).isEqualTo(Types.FixedType.ofLength(16));
}

@Test
void fixupShouldCorrectlyFixVariant() {
Copy link
Contributor

Choose a reason for hiding this comment

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

I don't think this behavior is correct. SparkFixupTypes should only fix the type when there is an explicit fix. But SparkFixupTypes doesn't have one so all types are converted to variant. I would expect this to only allow binary, which is probably what we will use to pass this value into Spark 3.5.

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Could you please further explain this? Do you mean that we need to restrict VariantType to only convert with BinaryType in SparkFixupTypes? Or are you just suggesting that the name of my test should not be "Correct"?

Copy link
Contributor

Choose a reason for hiding this comment

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

The cases that are fixed up are in SparkFixupTypes:

  @Override
  protected boolean fixupPrimitive(Type.PrimitiveType type, Type source) {
    switch (type.typeId()) {
      case STRING:
        if (source.typeId() == Type.TypeID.UUID) {
          return true;
        }
        break;
      case BINARY:
        if (source.typeId() == Type.TypeID.FIXED) {
          return true;
        }
        break;
      case TIMESTAMP:
        if (source.typeId() == Type.TypeID.TIMESTAMP) {
          return true;
        }
        break;
      default:
    }
    return false;
  }

Currently, the behavior is to always use the type from schema and never override with the reference schema's type. The test here validates that the type is not fixed, but what we care about are cases where a type does get fixed.

I think the case where a type does get fixed is when we are exposing variant as binary. So I would expect an update to SparkFixupTypes and the base FixupTypes to override binary with variant when variant is in the reference schema. Otherwise, the variant type is never "fixed" and we don't really need a new test suite.

Schema schema = new Schema(Types.NestedField.required(1, "field", Types.VariantType.get()));
Schema referenceSchema =
new Schema(Types.NestedField.required(1, "field", Types.StringType.get()));

Schema fixedSchema = SparkFixupTypes.fixup(schema, referenceSchema);

assertThat(fixedSchema.findType("field")).isSameAs(Types.VariantType.get());
}

@Test
void fixupShouldCorrectlyFixTimestamp() {
Schema schema =
new Schema(Types.NestedField.required(1, "field", Types.TimestampType.withZone()));
Schema referenceSchema =
new Schema(Types.NestedField.required(1, "field", Types.TimestampType.withoutZone()));

Schema fixedSchema = SparkFixupTypes.fixup(schema, referenceSchema);

assertThat(fixedSchema.findType("field")).isSameAs(Types.TimestampType.withoutZone());
}

@Test
void fixupShouldHandleNestedStructs() {
Schema schema =
new Schema(
Types.NestedField.required(
1,
"struct",
Types.StructType.of(
Types.NestedField.required(2, "field", Types.StringType.get()))));
Schema referenceSchema =
new Schema(
Types.NestedField.required(
1,
"struct",
Types.StructType.of(Types.NestedField.required(2, "field", Types.UUIDType.get()))));

Schema fixedSchema = SparkFixupTypes.fixup(schema, referenceSchema);

assertThat(fixedSchema.findType("struct"))
.isEqualTo(
Types.StructType.of(Types.NestedField.required(2, "field", Types.UUIDType.get())));
}

@Test
void fixupShouldHandleLists() {
Schema schema =
new Schema(
Types.NestedField.required(
1, "list", Types.ListType.ofRequired(2, Types.StringType.get())));
Schema referenceSchema =
new Schema(
Types.NestedField.required(
1, "list", Types.ListType.ofRequired(2, Types.UUIDType.get())));

Schema fixedSchema = SparkFixupTypes.fixup(schema, referenceSchema);

assertThat(fixedSchema.findType("list"))
.isEqualTo(Types.ListType.ofRequired(2, Types.UUIDType.get()));
}

@Test
void fixupShouldHandleMaps() {
Schema schema =
new Schema(
Types.NestedField.required(
1,
"map",
Types.MapType.ofRequired(2, 3, Types.StringType.get(), Types.StringType.get())));
Schema referenceSchema =
new Schema(
Types.NestedField.required(
1,
"map",
Types.MapType.ofRequired(2, 3, Types.UUIDType.get(), Types.UUIDType.get())));

Schema fixedSchema = SparkFixupTypes.fixup(schema, referenceSchema);

assertThat(fixedSchema.findType("map"))
.isEqualTo(Types.MapType.ofRequired(2, 3, Types.UUIDType.get(), Types.UUIDType.get()));
}
}