Skip to content

Spark 3.4: IcebergSource extends SessionConfigSupport #7732

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

Merged
szehon-ho merged 10 commits into from
Nov 23, 2024
Merged

Spark 3.4: IcebergSource extends SessionConfigSupport #7732

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
b21335a
Spark 3.4: IcebergSource extends SessionConfigSupport
pan3793 May 29, 2023
2dddaa7
tune
pan3793 Nov 18, 2024
9558bad
improve test
pan3793 Nov 19, 2024
f9d8ca4
style
pan3793 Nov 19, 2024
d3b9f9d
docs
pan3793 Nov 19, 2024
ea7a515
nit
pan3793 Nov 21, 2024
0b5ded7
Revert "nit"
pan3793 Nov 22, 2024
82bc24b
Revert "docs"
pan3793 Nov 22, 2024
5ee5cbf
improve assertion
pan3793 Nov 22, 2024
ff515b2
nit
pan3793 Nov 22, 2024
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
9 changes: 8 additions & 1 deletion spark/v3.4/spark/src/main/java/org/apache/iceberg/spark/source/IcebergSource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,7 @@
import org.apache.spark.sql.connector.catalog.CatalogManager;
import org.apache.spark.sql.connector.catalog.CatalogPlugin;
import org.apache.spark.sql.connector.catalog.Identifier;
import org.apache.spark.sql.connector.catalog.SessionConfigSupport;
import org.apache.spark.sql.connector.catalog.SupportsCatalogOptions;
import org.apache.spark.sql.connector.catalog.Table;
import org.apache.spark.sql.connector.catalog.TableCatalog;
Expand All @@ -61,7 +62,8 @@
* <p>The above list is in order of priority. For example: a matching catalog will take priority
* over any namespace resolution.
*/
public class IcebergSource implements DataSourceRegister, SupportsCatalogOptions {
public class IcebergSource
Copy link
Member

@szehon-ho szehon-ho Nov 18, 2024

Choose a reason for hiding this comment

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

Another comment, is it now multiple ways to configure properties (including #4011), it may be confusing to user. Worth to add a documentation about it, listing the precedence, ie:

I guess using dataframe API (to be double-checked)

  • explicit dataframe option
  • dataframe session default
  • if table exists, explicit table option
  • if table exists, table default

Copy link
Member Author

@pan3793 pan3793 Nov 19, 2024

Choose a reason for hiding this comment

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

I updated the docs and hope it's clear now.

implements DataSourceRegister, SupportsCatalogOptions, SessionConfigSupport {
private static final String DEFAULT_CATALOG_NAME = "default_iceberg";
private static final String DEFAULT_CACHE_CATALOG_NAME = "default_cache_iceberg";
private static final String DEFAULT_CATALOG = "spark.sql.catalog." + DEFAULT_CATALOG_NAME;
Expand All @@ -80,6 +82,11 @@ public String shortName() {
return "iceberg";
}

@Override
public String keyPrefix() {
return shortName();
}

@Override
public StructType inferSchema(CaseInsensitiveStringMap options) {
return null;
Expand Down
46 changes: 46 additions & 0 deletions ...v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestIcebergSourceTablesBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2184,6 +2184,52 @@ private void testWithFilter(String filterExpr, TableIdentifier tableIdentifier)
assertThat(actual).as("Rows must match").containsExactlyInAnyOrderElementsOf(expected);
}

@Test
public void testSessionConfigSupport() {
PartitionSpec spec = PartitionSpec.builderFor(SCHEMA).identity("id").build();
TableIdentifier tableIdentifier = TableIdentifier.of("db", "session_config_table");
Table table = createTable(tableIdentifier, SCHEMA, spec);

List<SimpleRecord> initialRecords =
Lists.newArrayList(
new SimpleRecord(1, "a"), new SimpleRecord(2, "b"), new SimpleRecord(3, "c"));

Dataset<Row> df = spark.createDataFrame(initialRecords, SimpleRecord.class);

df.select("id", "data")
.write()
.format("iceberg")
.mode(SaveMode.Append)
.save(loadLocation(tableIdentifier));

long s1 = table.currentSnapshot().snapshotId();

withSQLConf(
// set write option through session configuration
ImmutableMap.of("spark.datasource.iceberg.snapshot-property.foo", "bar"),
() -> {
df.select("id", "data")
.write()
.format("iceberg")
.mode(SaveMode.Append)
.save(loadLocation(tableIdentifier));
});

table.refresh();
assertThat(table.currentSnapshot().summary()).containsEntry("foo", "bar");

withSQLConf(
// set read option through session configuration
ImmutableMap.of("spark.datasource.iceberg.snapshot-id", String.valueOf(s1)),
() -> {
Dataset<Row> result = spark.read().format("iceberg").load(loadLocation(tableIdentifier));
List<SimpleRecord> actual = result.as(Encoders.bean(SimpleRecord.class)).collectAsList();
assertThat(actual)
.as("Rows must match")
.containsExactlyInAnyOrderElementsOf(initialRecords);
});
}

private GenericData.Record manifestRecord(
Table manifestTable, Long referenceSnapshotId, ManifestFile manifest) {
GenericRecordBuilder builder =
Expand Down