Spark: Provide Procedure for Catalog Register Procedure by RussellSpitzer · Pull Request #4810 · apache/iceberg

RussellSpitzer · 2022-05-18T15:49:55Z

Adds the ability to invoke a Catalog's register method via a Spark
procedure. This allows a user to create a catalog entry for a metadata.json
file which already exists but does not have a corrosponding catalog identifier.

Adds the ability to invoke a Catalog's register method via a Spark procedure. This allows a user to create a catalog entry for a metadata.json file which already exists but does not have a corrosponding catalog identifier.

RussellSpitzer · 2022-05-18T15:50:34Z

@karuppayya Could you please take a look?
@flyrain This is the API we were discussing

RussellSpitzer · 2022-05-18T15:51:35Z

+   * @param catalogName The name of the Spark Catalog being referenced
+   * @return the Iceberg catalog class being wrapped by the Spark Catalog
+   */
+  public static Catalog loadIcebergCatalog(SparkSession spark, String catalogName) {


This isn't strictly needed for this PR but is another API which is now possible because of this PR

@kbendick I think you would be interested in this as well

Thanks for tagging me @RussellSpitzer. I do think this would be really beneficial. I have a number of awkward workarounds that I've either suggsted to people or implemented myself that this would get around. So emphatic +1.

We'd probably want to add the HasIcebergCatalog interface to other catalogs as well, though that can be handled in a follow up once the interface is in.

If / when we merge this, let's open an issue for adding HasIcebergCatalog to other catalogs. I've got a newer way of tracking opened issues I've been using so that's one way to ensure that it is at least not forgotten about =)

One issue with that is this is a Spark Table Catalog implementing class in the current iteration, should we change that?

kbendick · 2022-05-18T17:20:58Z

+import org.apache.iceberg.catalog.Catalog;
+import org.apache.spark.sql.connector.catalog.TableCatalog;
+
+public interface HasIcebergCatalog extends TableCatalog {


👍 . This is something I get asked about often enough and I don't like having to ask end users to do multiple casts.

As a note, The reason I put this here as a public interface is that I didn't want to expose BaseCatalog

flyrain · 2022-05-18T19:57:26Z

+    Preconditions.checkArgument(metadataFile != null && !metadataFile.isEmpty(),
+        "Cannot handle an empty argument metadata_file");
+
+    Catalog icebergCatalog = ((HasIcebergCatalog) tableCatalog()).icebergCatalog();


Should we check if the tableCatalog is an instance of HasIcebergCatalog?

flyrain · 2022-05-18T19:59:08Z

+
+  @Override
+  public String description() {
+    return "MigrateTableProcedure";


MigrateTableProcedure -> RegisterTableProcedure ?

flyrain · 2022-05-18T20:04:21Z

+    long totalManifests = spark().table(tableCatalog().name() + "." + tableName + ".manifests").count();
+    long totalDataFiles = spark().table(tableCatalog().name() + "." + tableName + ".files").count();


I think this will scan 2 metadata tables, right? The concern is that the perf of the procedure may not be good if the table has a lot of data/metadata files. Can we use the ones from snapshot summary? e.g. TOTAL_DATA_FILES_PROP

I kinda wanted to do a little sanity check here that the table actually worked which is why I had the full scans here

flyrain · 2022-05-18T20:12:20Z

+  public static Catalog loadIcebergCatalog(SparkSession spark, String catalogName) {
+    CatalogPlugin catalogPlugin = spark.sessionState().catalogManager().catalog(catalogName);
+    Preconditions.checkArgument(catalogPlugin instanceof HasIcebergCatalog,
+        String.format("Cannot load Iceberg catalog from catalog %s because it is not a Spark Iceberg catalog. Actual " +


because it is not a Spark Iceberg catalog -> because it doesn't contain an Iceberg catalog?

RussellSpitzer · 2022-05-19T19:23:38Z

Thanks @flyrain and @kbendick , Please take a look again when you have a chance

flyrain · 2022-05-19T23:16:14Z

+
+    Catalog icebergCatalog = ((HasIcebergCatalog) tableCatalog()).icebergCatalog();
+    Table table = icebergCatalog.registerTable(tableName, metadataFile);
+    Snapshot currentSnapshot = table.currentSnapshot();


I think we need a null check here. There is an edge case, that current snapshot is null.

flyrain

Thanks for the PR. LGTM overall. The only thing is the null check of the current snapshot.

flyrain

+1

singhpk234

LGTM as well, Thanks @RussellSpitzer :) !!!

singhpk234 · 2022-05-27T17:33:54Z

+    assertEquals("Registered table rows should match original table rows", original, registered);
+    Assert.assertEquals("Should have the right row count in the procedure result",
+        numRows, result.get(0)[1]);
+    Assert.assertEquals("Should have the right datafile count the procedure result",


nit

Suggested change

Assert.assertEquals("Should have the right datafile count the procedure result",

Assert.assertEquals("Should have the right datafile count in the procedure result",

singhpk234 · 2022-05-27T17:38:05Z

+  };
+
+  private static final StructType OUTPUT_TYPE = new StructType(new StructField[]{
+      new StructField("Current Snapshot", DataTypes.LongType, true, Metadata.empty()),


[minor] should we call this Current SnapshotId

kbendick

+1 this LGTM as well. Are we still interested in merging this @RussellSpitzer?

RussellSpitzer · 2022-05-28T19:50:19Z

Yes sorry the merge alignment pr I just merged in was high priority and this went on my back burner

RussellSpitzer · 2022-06-03T13:10:11Z

Thanks @flyrain, @kbendick and @singhpk234 !

* Spark: Provide Procedure for Catalog Register Procedure Adds the ability to invoke a Catalog's register method via a Spark procedure. This allows a user to create a catalog entry for a metadata.json file which already exists but does not have a corresponding catalog identifier.

Spark: Provide Procedure for Catalog Register Procedure

f4831e4

Adds the ability to invoke a Catalog's register method via a Spark procedure. This allows a user to create a catalog entry for a metadata.json file which already exists but does not have a corrosponding catalog identifier.

RussellSpitzer commented May 18, 2022

View reviewed changes

RussellSpitzer added 2 commits May 18, 2022 10:52

Reword Util Error Message

de9449d

Swap subclasses for Interface

92dc1d2

github-actions Bot added core spark labels May 18, 2022

kbendick reviewed May 18, 2022

View reviewed changes

flyrain reviewed May 18, 2022

View reviewed changes

Reviewer comments

db533b6

flyrain reviewed May 19, 2022

View reviewed changes

Null check on Current Snapshot

dc719a4

flyrain approved these changes May 25, 2022

View reviewed changes

Add version.txt for release 0.13.2

268bab6

singhpk234 approved these changes May 27, 2022

View reviewed changes

kbendick approved these changes May 27, 2022

View reviewed changes

Reviewer Comments

9e51245

RussellSpitzer merged commit 82607f1 into apache:master Jun 3, 2022

RussellSpitzer deleted the registerTableProcedure branch June 3, 2022 13:09

singhpk234 mentioned this pull request Jun 9, 2022

Spark: Show metadata.json path in DESC TABLE EXTENDED #5006

Closed

RussellSpitzer mentioned this pull request Jun 29, 2022

Support catalog method to set table metadata #5163

Closed

rajarshisarkar mentioned this pull request Dec 7, 2022

Docs: Add register table Spark procedure documentation #6376

Merged

		long totalManifests = spark().table(tableCatalog().name() + "." + tableName + ".manifests").count();
		long totalDataFiles = spark().table(tableCatalog().name() + "." + tableName + ".files").count();

	Assert.assertEquals("Should have the right datafile count the procedure result",
	Assert.assertEquals("Should have the right datafile count in the procedure result",

Uh oh!

Conversation

RussellSpitzer commented May 18, 2022

Uh oh!

RussellSpitzer commented May 18, 2022

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

flyrain May 18, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

RussellSpitzer commented May 19, 2022

Uh oh!

Choose a reason for hiding this comment

Uh oh!

flyrain left a comment

Choose a reason for hiding this comment

Uh oh!

flyrain left a comment

Choose a reason for hiding this comment

Uh oh!

singhpk234 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kbendick left a comment

Choose a reason for hiding this comment

Uh oh!

RussellSpitzer commented May 28, 2022

Uh oh!

RussellSpitzer commented Jun 3, 2022

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

flyrain May 18, 2022 •

edited

Loading