Hive: Add View support for HIVE catalog

[nk1506]

This changes include:

1. Introduction of common metadata interface(BaseMetadata) for table and view.
2. Refactor for HiveTableOperation to have common code for table and view commits.

Ref: #9682 , #9461 , #8907

[szehon-ho]

Thanks for combining the earlier prs. It's clearer now.

Some preliminary comments.

@nastra please take a look as well to see if there are any major concerns.

[nastra]

I actually think it's missing calling commit here where it should be expecting the exception that is being set up in failCommitAndThrowException.

Can you please update the test to

@Test
  public void testThriftExceptionUnknownStateIfNotInHistoryFailureOnCommit()
      throws TException, InterruptedException {
    HiveViewOperations ops = (HiveViewOperations) ((BaseView) view).operations();
    ViewMetadata metadataV1 = ops.current();
    assertThat(metadataV1.properties()).hasSize(0);

    view.updateProperties().set("k1", "v1").commit();
    ops.refresh();
    ViewMetadata metadataV2 = ops.current();
    assertThat(metadataV2.properties()).hasSize(1).containsEntry("k1", "v1");

    HiveViewOperations spyOps = spy(ops);

    failCommitAndThrowException(spyOps);

    assertThatThrownBy(() -> spyOps.commit(metadataV2, metadataV1))
            .isInstanceOf(CommitStateUnknownException.class)
            .hasMessageStartingWith("Datacenter on fire");

    ops.refresh();

    assertThat(ops.current()).as("Current metadata should not have changed").isEqualTo(metadataV2);
    assertThat(metadataFileExists(metadataV2)).as("Current metadata should still exist").isTrue();
    assertThat(metadataFileCount(ops.current()))
        .as(
            "New metadata files should still exist, new location not in history but"
                + " the commit may still succeed")
        .isEqualTo(2);
  }

[nastra]

please update this one to

@Test
  public void testThriftExceptionSuccessOnCommit() throws TException, InterruptedException {
    HiveViewOperations ops = (HiveViewOperations) ((BaseView) view).operations();

    ViewMetadata metadataV1 = ops.current();
    assertThat(metadataV1.properties()).hasSize(0);

    view.updateProperties().set("k1", "v1").commit();
    ops.refresh();
    ViewMetadata metadataV2 = ops.current();
    assertThat(metadataV2.properties()).hasSize(1).containsEntry("k1", "v1");

    HiveViewOperations spyOps = spy(ops);

    // Simulate a communication error after a successful commit
    commitAndThrowException(ops, spyOps);
    // Shouldn't throw because the commit succeeds even though an exception is thrown
    spyOps.commit(metadataV2, metadataV1);

    assertThat(ops.current()).as("Current metadata should have not changed").isEqualTo(metadataV2);
    assertThat(metadataFileExists(ops.current()))
        .as("Current metadata file should still exist")
        .isTrue();
  }

[nk1506]

Build is failing because of flaky issue #11046

[nastra]

will wait a bit with merging in case @danielcweeks has some additional comments

[nastra]

thanks @nk1506 for your patience here. Also thanks for everyone that helped out with reviews

[nk1506]

Thanks @nastra and @danielcweeks for your reviews and feedback.

[nqvuong1998]

Hi @nastra @danielcweeks , Is this pull request scheduled for inclusion in the v1.7 release?

[nastra]

@nqvuong1998 yes that will be shipped with 1.7.0