[RTAS]: Fix bug - remove fs scheme from tableLocation in commit (cont) (#594)

## Summary
This is the follow up PR of
#542. After the fix, we were
able to replace the table once but not multiple times. RC anslysis:
1. `OpenHouseInternalTableOperations` uses `metadata.location()` for the
new HTS tableLocation, and that value comes from
`.withLocation(tableLocation)` that we pass, so we need to make sure
it's schemeless.
2. We used `tableDto.getTableVersion()` to populate
`.withLocation(tableLocation)`, but tableDto always contains scheme.
(Why is tableLocation = tableVersion in tableDto? Because tableVersion
comes from client request, and client request uses the tableLocation
from the server response, so they will always be same and contain
scheme)
3. The last PR fixes the tableLocation in table properties, so the new
tableVersion for the first replace will be schemeless. But since the new
tableLocation has scheme, the next replace will fail.

Therefore, in this PR, we use the tableLocation from the table
properties that we had just stripped in the last PR for
`.withLocation(tableLocation)`.

## Changes

- [ ] Client-facing API Changes
- [ ] Internal API Changes
- [x] Bug Fixes
- [ ] New Features
- [ ] Performance Improvements
- [ ] Code Style
- [ ] Refactoring
- [ ] Documentation
- [ ] Tests

For all the boxes checked, please include additional details of the
changes made in this pull request.

## Testing Done
<!--- Check any relevant boxes with "x" -->

- [x] Manually Tested on local docker setup. Please include commands
ran, and their output.
- [ ] Added new tests for the changes made.
- [x] Updated existing tests to reflect the changes made.
- [ ] No tests added or updated. Please explain why. If unsure, please
feel free to ask for help.
- [ ] Some other form of testing like staging or soak time in
production. Please explain.

**Unit tests**
I removed `stripPathScheme` in the spark E2E tests to gurantee that
scheme issues can be captured by unit tests.

**Test 1: RTAS**
```
scala> spark.sql(s"CREATE TABLE $tableName TBLPROPERTIES ('prop1'='val1', 'prop2'='val2') AS SELECT * FROM $sourceName");
res3: org.apache.spark.sql.DataFrame = []                                       

scala> spark.sql(s"INSERT INTO $tableName values (4, 'd')");
res4: org.apache.spark.sql.DataFrame = []

scala> spark.sql(s"REPLACE TABLE $tableName PARTITIONED BY (part) TBLPROPERTIES ('prop1'='newval1', 'prop3'='val3') AS SELECT id, data, CASE WHEN (id % 2) = 0 THEN 'even' ELSE 'odd' END AS part FROM $sourceName ORDER BY 3, 1");
res5: org.apache.spark.sql.DataFrame = []

scala> spark.sql(s"REPLACE TABLE $tableName PARTITIONED BY (part) AS SELECT 2 * id as id, data, CASE WHEN ((2 * id) % 2) = 0 THEN 'even' ELSE 'odd' END AS part FROM $sourceName ORDER BY 3, 1");
res12: org.apache.spark.sql.DataFrame = []

scala> spark.sql(s"SELECT * FROM $tableName").show(false)
+---+----+----+
|id |data|part|
+---+----+----+
|2  |a   |even|
|4  |b   |even|
|6  |c   |even|
+---+----+----+
```
**Test 2: CRTAS**
```
scala> spark.sql(s"CREATE OR REPLACE TABLE $tableName TBLPROPERTIES ('prop1'='val1', 'prop2'='val2') AS SELECT * FROM $sourceName");
res19: org.apache.spark.sql.DataFrame = []

scala> spark.sql(s"INSERT INTO $tableName values (4, 'd')");
res20: org.apache.spark.sql.DataFrame = []

scala> spark.sql(s"CREATE OR REPLACE TABLE $tableName PARTITIONED BY (part) AS SELECT id, data, CASE WHEN id % 2 = 0 THEN 'even' ELSE 'odd' END AS part FROM $sourceName ORDER BY 3, 1");
res21: org.apache.spark.sql.DataFrame = []

scala> spark.sql(s"CREATE OR REPLACE TABLE $tableName PARTITIONED BY (part) AS SELECT 2 * id as id, data, CASE WHEN ((2 * id) % 2) = 0 THEN 'even' ELSE 'odd' END AS part FROM $sourceName ORDER BY 3, 1");
res22: org.apache.spark.sql.DataFrame = []

scala> spark.sql(s"SELECT * FROM $tableName").show(false)
+---+----+----+
|id |data|part|
+---+----+----+
|2  |a   |even|
|4  |b   |even|
|6  |c   |even|
+---+----+----+
```
# Additional Information

- [ ] Breaking Changes
- [ ] Deprecations
- [ ] Large PR broken into smaller PRs, and PR plan linked in the
description.

For all the boxes checked, include additional details of the changes
made in this pull request.

---------

Co-authored-by: Levi Jiang <lejiang@lejiang-mn2962.linkedin.biz>

Author: jiang95-dev

apps/spark/src/test/java/com/linkedin/openhouse/catalog/e2e/RTASJavaTest.java

@@ -141,8 +141,8 @@ private void verifyReplacedTable(
     Table replacedTable = catalog.loadTable(TABLE_IDENT);
 
     assertEquals(
-        stripPathScheme(originalLocation),
-        stripPathScheme(replacedTable.location()),
+        originalLocation,
+        replacedTable.location(),
         "Table location should be preserved after replace");
     assertEquals(
         REPLACE_SCHEMA.asStruct(),

integrations/spark/spark-3.1/openhouse-spark-itest/src/test/java/com/linkedin/openhouse/spark/catalogtest/RTASTest.java

@@ -83,10 +83,7 @@ public void testRTAS() throws Exception {
       Table rtasTable = catalog.loadTable(tableIdent);
 
       // verify table location is unchanged
-      assertEquals(
-          stripPathScheme(expectedTableLocation),
-          stripPathScheme(rtasTable.location()),
-          "Should have same table location");
+      assertEquals(expectedTableLocation, rtasTable.location(), "Should have same table location");
       // verify schema and spec are changed
       assertEquals(
           expectedSchema.asStruct(),
@@ -151,10 +148,7 @@ public void testCreateRTAS() throws Exception {
       Table rtasTable = catalog.loadTable(tableIdent);
 
       // verify table location is unchanged
-      assertEquals(
-          stripPathScheme(expectedTableLocation),
-          stripPathScheme(rtasTable.location()),
-          "Should have same table location");
+      assertEquals(expectedTableLocation, rtasTable.location(), "Should have same table location");
       // verify schema and spec are changed
       assertEquals(
           expectedSchema.asStruct(),
@@ -201,10 +195,7 @@ public void testDataFrameV2Replace() throws Exception {
       Table rtasTable = catalog.loadTable(tableIdent);
 
       // verify table location is unchanged
-      assertEquals(
-          stripPathScheme(expectedTableLocation),
-          stripPathScheme(rtasTable.location()),
-          "Should have same table location");
+      assertEquals(expectedTableLocation, rtasTable.location(), "Should have same table location");
       // verify schema and spec are changed
       assertEquals(
           expectedSchema.asStruct(),
@@ -265,10 +256,7 @@ public void testDataFrameV2CreateOrReplace() throws Exception {
       Table rtasTable = catalog.loadTable(tableIdent);
 
       // verify table location is unchanged
-      assertEquals(
-          stripPathScheme(expectedTableLocation),
-          stripPathScheme(rtasTable.location()),
-          "Should have same table location");
+      assertEquals(expectedTableLocation, rtasTable.location(), "Should have same table location");
       // verify schema and spec are changed
       assertEquals(
           expectedSchema.asStruct(),

services/tables/src/main/java/com/linkedin/openhouse/tables/repository/impl/OpenHouseInternalRepositoryImpl.java

@@ -2,6 +2,7 @@
 
 import static com.linkedin.openhouse.internal.catalog.CatalogConstants.*;
 import static com.linkedin.openhouse.internal.catalog.mapper.HouseTableSerdeUtils.*;
+import static com.linkedin.openhouse.internal.catalog.mapper.HouseTableSerdeUtils.getCanonicalFieldName;
 import static com.linkedin.openhouse.tables.repository.impl.InternalRepositoryUtils.*;
 
 import com.google.common.annotations.VisibleForTesting;
@@ -158,8 +159,8 @@ public TableDto save(TableDto tableDto) {
       Map<String, String> tableProps = computePropsForTableCreation(tableDto);
       tablePolicyManager.managePoliciesOnCreateIfNeeded(tableDto);
       SortOrder sortOrder = getIcebergSortOrder(tableDto, writeSchema);
-      String tableLocation =
-          tableDto.getTableVersion().substring(0, tableDto.getTableVersion().lastIndexOf("/"));
+      String metadataLocation = tableProps.get(getCanonicalFieldName("tableLocation"));
+      String tableLocation = metadataLocation.substring(0, metadataLocation.lastIndexOf("/"));
       table =
           replaceTable(
               tableIdentifier, writeSchema, partitionSpec, tableLocation, tableProps, sortOrder);

tables-test-fixtures/tables-test-fixtures-iceberg-1.2/src/main/java/com/linkedin/openhouse/tablestest/OpenHouseSparkITest.java

@@ -131,11 +131,4 @@ protected Catalog getOpenHouseCatalog(SparkSession spark) {
         catalogProperties,
         spark.sparkContext().hadoopConfiguration());
   }
-
-  /**
-   * Getting rid of "file:" part if needed for ease of comparison of tableLocation / tableVersion
-   */
-  protected String stripPathScheme(String path) {
-    return path.startsWith("file:") ? path.split("file:")[1] : path;
-  }
 }