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: Correct partition transform functions to match spec.md #8192

Merged
merged 12 commits into from
Sep 25, 2023
Merged

Spark: Correct partition transform functions to match spec.md #8192

Show file tree
Hide file tree
Changes from 9 commits
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
1d6776d
add non-plural transform function names
clettieri Aug 1, 2023
c20e700
typos
clettieri Aug 1, 2023
d95d4c3
typo
clettieri Aug 1, 2023
ef6a352
add tests for singular partition transform functions
clettieri Aug 1, 2023
deef41a
update tests to assertj
clettieri Sep 13, 2023
9a62170
add spark 3.5
clettieri Sep 20, 2023
a886a6a
use new createTable() in tests
clettieri Sep 20, 2023
0b9c76e
formatting
clettieri Sep 20, 2023
cc49a0a
create table test
clettieri Sep 21, 2023
c13c989
Add missing : to fix syntax error.
rdblue Sep 24, 2023
3bcdd3d
Apply spotless.
rdblue Sep 24, 2023
60b77f6
Merge pull request #1 from rdblue/docs-fix-partition-transforms
clettieri Sep 25, 2023
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
85 changes: 85 additions & 0 deletions ...ions/src/test/java/org/apache/iceberg/spark/extensions/TestAlterTablePartitionFields.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@
import org.junit.Assert;
import org.junit.Test;
import org.junit.runners.Parameterized;
import org.assertj.core.api.Assertions;

public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {

Expand Down Expand Up @@ -179,6 +180,90 @@ public void testAddHoursPartition() {
Assert.assertEquals("Should have new spec field", expected, table.spec());
}

@Test
public void testAddYearPartition() {
createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);

Assertions.assertThat(table.spec().isUnpartitioned())
.as("Table should start unpartitioned")
.isTrue();

sql("ALTER TABLE %s ADD PARTITION FIELD year(ts)", tableName);

table.refresh();

PartitionSpec expected =
PartitionSpec.builderFor(table.schema()).withSpecId(1).year("ts").build();

Assertions.assertThat(table.spec())
.as("Should have new spec field")
.isEqualTo(expected);
}

@Test
public void testAddMonthPartition() {
createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);

Assertions.assertThat(table.spec().isUnpartitioned())
.as("Table should start unpartitioned")
.isTrue();

sql("ALTER TABLE %s ADD PARTITION FIELD month(ts)", tableName);

table.refresh();

PartitionSpec expected =
PartitionSpec.builderFor(table.schema()).withSpecId(1).month("ts").build();

Assertions.assertThat(table.spec())
.as("Should have new spec field")
.isEqualTo(expected);
}

@Test
public void testAddDayPartition() {
createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);

Assertions.assertThat(table.spec().isUnpartitioned())
.as("Table should start unpartitioned")
.isTrue();

sql("ALTER TABLE %s ADD PARTITION FIELD day(ts)", tableName);

table.refresh();

PartitionSpec expected =
PartitionSpec.builderFor(table.schema()).withSpecId(1).day("ts").build();

Assertions.assertThat(table.spec())
.as("Should have new spec field")
.isEqualTo(expected);
}

@Test
public void testAddHourPartition() {
createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);

Assertions.assertThat(table.spec().isUnpartitioned())
.as("Table should start unpartitioned")
.isTrue();

sql("ALTER TABLE %s ADD PARTITION FIELD hour(ts)", tableName);

table.refresh();

PartitionSpec expected =
PartitionSpec.builderFor(table.schema()).withSpecId(1).hour("ts").build();

Assertions.assertThat(table.spec())
.as("Should have new spec field")
.isEqualTo(expected);
}

@Test
public void testAddNamedPartition() {
createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Expand Down
8 changes: 8 additions & 0 deletions spark/v3.1/spark/src/main/java/org/apache/iceberg/spark/Spark3Util.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -374,14 +374,18 @@ public static Term toIcebergTerm(Transform transform) {
return org.apache.iceberg.expressions.Expressions.ref(colName);
case "bucket":
return org.apache.iceberg.expressions.Expressions.bucket(colName, findWidth(transform));
case "year":
case "years":
return org.apache.iceberg.expressions.Expressions.year(colName);
case "month":
case "months":
return org.apache.iceberg.expressions.Expressions.month(colName);
case "date":
case "day":
case "days":
return org.apache.iceberg.expressions.Expressions.day(colName);
case "date_hour":
case "hour":
case "hours":
return org.apache.iceberg.expressions.Expressions.hour(colName);
case "truncate":
Expand Down Expand Up @@ -417,17 +421,21 @@ public static PartitionSpec toPartitionSpec(Schema schema, Transform[] partition
case "bucket":
builder.bucket(colName, findWidth(transform));
break;
case "year":
case "years":
builder.year(colName);
break;
case "month":
case "months":
builder.month(colName);
break;
case "date":
case "day":
case "days":
builder.day(colName);
break;
case "date_hour":
case "hour":
case "hours":
builder.hour(colName);
break;
Expand Down
20 changes: 20 additions & 0 deletions spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/sql/TestCreateTable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,26 @@ public void testTransformIgnoreCase() {
Assert.assertTrue("Table should already exist", validationCatalog.tableExists(tableIdent));
}

@Test
public void testTransformSingularForm() {
Assert.assertFalse("Table should not already exist", validationCatalog.tableExists(tableIdent));
sql(
"CREATE TABLE IF NOT EXISTS %s (id BIGINT NOT NULL, ts timestamp) "
+ "USING iceberg partitioned by (hour(ts))",
tableName);
Assert.assertTrue("Table should exist", validationCatalog.tableExists(tableIdent));
}

@Test
public void testTransformPluralForm() {
Assert.assertFalse("Table should not already exist", validationCatalog.tableExists(tableIdent));
sql(
"CREATE TABLE IF NOT EXISTS %s (id BIGINT NOT NULL, ts timestamp) "
+ "USING iceberg partitioned by (hours(ts))",
tableName);
Assert.assertTrue("Table should exist", validationCatalog.tableExists(tableIdent));
}

@Test
public void testCreateTable() {
Assert.assertFalse("Table should not already exist", validationCatalog.tableExists(tableIdent));
Expand Down
85 changes: 85 additions & 0 deletions ...ions/src/test/java/org/apache/iceberg/spark/extensions/TestAlterTablePartitionFields.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@
import org.junit.Assert;
import org.junit.Test;
import org.junit.runners.Parameterized;
import org.assertj.core.api.Assertions;

public class TestAlterTablePartitionFields extends SparkExtensionsTestBase {

Expand Down Expand Up @@ -179,6 +180,90 @@ public void testAddHoursPartition() {
Assert.assertEquals("Should have new spec field", expected, table.spec());
}

@Test
public void testAddYearPartition() {
createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);

Assertions.assertThat(table.spec().isUnpartitioned())
.as("Table should start unpartitioned")
.isTrue();

sql("ALTER TABLE %s ADD PARTITION FIELD year(ts)", tableName);

table.refresh();

PartitionSpec expected =
PartitionSpec.builderFor(table.schema()).withSpecId(1).year("ts").build();

Assertions.assertThat(table.spec())
.as("Should have new spec field")
.isEqualTo(expected);
}

@Test
public void testAddMonthPartition() {
createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);

Assertions.assertThat(table.spec().isUnpartitioned())
.as("Table should start unpartitioned")
.isTrue();

sql("ALTER TABLE %s ADD PARTITION FIELD month(ts)", tableName);

table.refresh();

PartitionSpec expected =
PartitionSpec.builderFor(table.schema()).withSpecId(1).month("ts").build();

Assertions.assertThat(table.spec())
.as("Should have new spec field")
.isEqualTo(expected);
}

@Test
public void testAddDayPartition() {
createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);

Assertions.assertThat(table.spec().isUnpartitioned())
.as("Table should start unpartitioned")
.isTrue();

sql("ALTER TABLE %s ADD PARTITION FIELD day(ts)", tableName);

table.refresh();

PartitionSpec expected =
PartitionSpec.builderFor(table.schema()).withSpecId(1).day("ts").build();

Assertions.assertThat(table.spec())
.as("Should have new spec field")
.isEqualTo(expected);
}

@Test
public void testAddHourPartition() {
createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Table table = validationCatalog.loadTable(tableIdent);

Assertions.assertThat(table.spec().isUnpartitioned())
.as("Table should start unpartitioned")
.isTrue();

sql("ALTER TABLE %s ADD PARTITION FIELD hour(ts)", tableName);

table.refresh();

PartitionSpec expected =
PartitionSpec.builderFor(table.schema()).withSpecId(1).hour("ts").build();

Assertions.assertThat(table.spec())
.as("Should have new spec field")
.isEqualTo(expected);
}

@Test
public void testAddNamedPartition() {
createTable("id bigint NOT NULL, category string, ts timestamp, data string");
Expand Down
8 changes: 8 additions & 0 deletions spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/Spark3Util.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -342,14 +342,18 @@ public static Term toIcebergTerm(Expression expr) {
return org.apache.iceberg.expressions.Expressions.ref(colName);
case "bucket":
return org.apache.iceberg.expressions.Expressions.bucket(colName, findWidth(transform));
case "year":
case "years":
return org.apache.iceberg.expressions.Expressions.year(colName);
case "month":
case "months":
return org.apache.iceberg.expressions.Expressions.month(colName);
case "date":
case "day":
case "days":
return org.apache.iceberg.expressions.Expressions.day(colName);
case "date_hour":
case "hour":
case "hours":
return org.apache.iceberg.expressions.Expressions.hour(colName);
case "truncate":
Expand Down Expand Up @@ -399,17 +403,21 @@ public static PartitionSpec toPartitionSpec(Schema schema, Transform[] partition
case "bucket":
builder.bucket(colName, findWidth(transform));
break;
case "year":
case "years":
builder.year(colName);
break;
case "month":
case "months":
builder.month(colName);
break;
case "date":
case "day":
case "days":
builder.day(colName);
break;
case "date_hour":
case "hour":
case "hours":
builder.hour(colName);
break;
Expand Down
20 changes: 20 additions & 0 deletions spark/v3.2/spark/src/test/java/org/apache/iceberg/spark/sql/TestCreateTable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,26 @@ public void testTransformIgnoreCase() {
Assert.assertTrue("Table should already exist", validationCatalog.tableExists(tableIdent));
}

@Test
public void testTransformSingularForm() {
Assert.assertFalse("Table should not already exist", validationCatalog.tableExists(tableIdent));
sql(
"CREATE TABLE IF NOT EXISTS %s (id BIGINT NOT NULL, ts timestamp) "
+ "USING iceberg partitioned by (hour(ts))",
tableName);
Assert.assertTrue("Table should exist", validationCatalog.tableExists(tableIdent));
}

@Test
public void testTransformPluralForm() {
Assert.assertFalse("Table should not already exist", validationCatalog.tableExists(tableIdent));
sql(
"CREATE TABLE IF NOT EXISTS %s (id BIGINT NOT NULL, ts timestamp) "
+ "USING iceberg partitioned by (hours(ts))",
tableName);
Assert.assertTrue("Table should exist", validationCatalog.tableExists(tableIdent));
}

@Test
public void testCreateTable() {
Assert.assertFalse("Table should not already exist", validationCatalog.tableExists(tableIdent));
Expand Down
Loading
Loading