This is an automated email from the ASF dual-hosted git repository.
jinsongzhou pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/amoro.git
The following commit(s) were added to refs/heads/master by this push:
new ba025277a [AMORO-2629]Mixed hive refractor (#2691)
ba025277a is described below
commit ba025277ae8554243f2809ca0ff890168e6e1d32
Author: Wang Tao <[email protected]>
AuthorDate: Fri Apr 12 12:03:58 2024 +0800
[AMORO-2629]Mixed hive refractor (#2691)
* issue_2629 mixed-hive refractor
* issue_2629 mixed-hive refractor
* issue_2629 mixed-hive refractor
* issue_2629 mixed-hive refractor
* issue_2629 mixed-hive refractor
---
.../arctic/server/catalog/CatalogBuilder.java | 4 -
.../server/catalog/MixedHiveCatalogImpl.java | 5 +-
.../dashboard/controller/TableController.java | 17 +-
.../arctic/server/table/AMSTableTestBase.java | 40 +-
.../server/table/TestTableRuntimeHandler.java | 7 +-
.../arctic/server/table/TestTableService.java | 29 +-
.../netease/arctic/catalog/BasicArcticCatalog.java | 20 +-
.../com/netease/arctic/catalog/MixedTables.java | 4 +
.../java/com/netease/arctic/hive/HMSClient.java | 2 +
.../com/netease/arctic/hive/HMSClientImpl.java | 6 +
.../arctic/hive/catalog/ArcticHiveCatalog.java | 461 +++++++++++++++++++--
.../arctic/hive/catalog/MixedHiveTables.java | 142 ++++++-
.../arctic/hive/catalog/HiveCatalogTestHelper.java | 5 +-
.../arctic/spark/ArcticSparkExtensions.scala | 7 +
.../arctic/spark/ArcticSparkExtensions.scala | 7 +
15 files changed, 681 insertions(+), 75 deletions(-)
diff --git
a/ams/server/src/main/java/com/netease/arctic/server/catalog/CatalogBuilder.java
b/ams/server/src/main/java/com/netease/arctic/server/catalog/CatalogBuilder.java
index 0eb6abd94..3f276a30b 100644
---
a/ams/server/src/main/java/com/netease/arctic/server/catalog/CatalogBuilder.java
+++
b/ams/server/src/main/java/com/netease/arctic/server/catalog/CatalogBuilder.java
@@ -82,10 +82,6 @@ public class CatalogBuilder {
case CATALOG_TYPE_CUSTOM:
return new ExternalCatalog(catalogMeta);
case CATALOG_TYPE_HIVE:
- if (tableFormats.size() == 1 &&
tableFormat.equals(TableFormat.MIXED_HIVE)) {
- return new MixedHiveCatalogImpl(catalogMeta);
- }
- // if tableFormats.size() > 1 , we need fulfill the ams uri in
catalogProperty
String amsUri = getAmsURI(serverConfiguration);
catalogMeta.getCatalogProperties().put(CatalogMetaProperties.AMS_URI,
amsUri);
return new ExternalCatalog(catalogMeta);
diff --git
a/ams/server/src/main/java/com/netease/arctic/server/catalog/MixedHiveCatalogImpl.java
b/ams/server/src/main/java/com/netease/arctic/server/catalog/MixedHiveCatalogImpl.java
index cda5f9190..dd004413d 100644
---
a/ams/server/src/main/java/com/netease/arctic/server/catalog/MixedHiveCatalogImpl.java
+++
b/ams/server/src/main/java/com/netease/arctic/server/catalog/MixedHiveCatalogImpl.java
@@ -21,7 +21,6 @@ package com.netease.arctic.server.catalog;
import com.netease.arctic.AmoroTable;
import com.netease.arctic.TableFormat;
import com.netease.arctic.api.CatalogMeta;
-import com.netease.arctic.catalog.MixedTables;
import com.netease.arctic.formats.mixed.MixedTable;
import com.netease.arctic.hive.CachedHiveClientPool;
import com.netease.arctic.hive.HMSClient;
@@ -38,7 +37,7 @@ import org.apache.thrift.TException;
import java.util.List;
public class MixedHiveCatalogImpl extends InternalCatalog {
- protected MixedTables tables;
+ protected MixedHiveTables tables;
private volatile CachedHiveClientPool hiveClientPool;
protected MixedHiveCatalogImpl(CatalogMeta catalogMeta) {
@@ -133,7 +132,7 @@ public class MixedHiveCatalogImpl extends InternalCatalog {
return hiveClientPool;
}
- private MixedTables tables() {
+ private MixedHiveTables tables() {
return tables;
}
}
diff --git
a/ams/server/src/main/java/com/netease/arctic/server/dashboard/controller/TableController.java
b/ams/server/src/main/java/com/netease/arctic/server/dashboard/controller/TableController.java
index d0ad2b3ac..78e136f89 100644
---
a/ams/server/src/main/java/com/netease/arctic/server/dashboard/controller/TableController.java
+++
b/ams/server/src/main/java/com/netease/arctic/server/dashboard/controller/TableController.java
@@ -199,13 +199,22 @@ public class TableController {
String amsUri = AmsUtil.getAMSThriftAddress(serviceConfig,
Constants.THRIFT_TABLE_SERVICE_NAME);
catalogMeta.putToCatalogProperties(CatalogMetaProperties.AMS_URI, amsUri);
TableMetaStore tableMetaStore =
ArcticCatalogUtil.buildMetaStore(catalogMeta);
+ // check whether catalog support MIXED_HIVE format.
+ Set<TableFormat> tableFormats =
ArcticCatalogUtil.tableFormats(catalogMeta);
+ Preconditions.checkState(
+ tableFormats.contains(TableFormat.MIXED_HIVE),
+ "Catalog %s does not support MIXED_HIVE format",
+ catalog);
+ // we should only keep MIXED_HIVE format，
+ // so `CatalogLoader.createCatalog` can get right CatalogImpl through
calling catalogImpl.
+ Map<String, String> originCatalogProperties =
catalogMeta.getCatalogProperties();
+ Map<String, String> catalogProperties = new
HashMap<>(originCatalogProperties);
+ catalogProperties.put(CatalogMetaProperties.TABLE_FORMATS,
TableFormat.MIXED_HIVE.name());
+
ArcticHiveCatalog arcticHiveCatalog =
(ArcticHiveCatalog)
CatalogLoader.createCatalog(
- catalog,
- catalogMeta.getCatalogType(),
- catalogMeta.getCatalogProperties(),
- tableMetaStore);
+ catalog, catalogMeta.getCatalogType(), catalogProperties,
tableMetaStore);
tableUpgradeExecutor.execute(
() -> {
diff --git
a/ams/server/src/test/java/com/netease/arctic/server/table/AMSTableTestBase.java
b/ams/server/src/test/java/com/netease/arctic/server/table/AMSTableTestBase.java
index b36f2fe0f..1bdb63c3a 100644
---
a/ams/server/src/test/java/com/netease/arctic/server/table/AMSTableTestBase.java
+++
b/ams/server/src/test/java/com/netease/arctic/server/table/AMSTableTestBase.java
@@ -30,6 +30,8 @@ import com.netease.arctic.catalog.CatalogLoader;
import com.netease.arctic.catalog.CatalogTestHelper;
import com.netease.arctic.catalog.MixedTables;
import com.netease.arctic.hive.TestHMS;
+import com.netease.arctic.hive.catalog.ArcticHiveCatalog;
+import com.netease.arctic.properties.CatalogMetaProperties;
import com.netease.arctic.table.ArcticTable;
import com.netease.arctic.utils.ArcticCatalogUtil;
import com.netease.arctic.utils.ConvertStructUtil;
@@ -47,6 +49,7 @@ import org.junit.Rule;
import org.junit.rules.TemporaryFolder;
import java.io.IOException;
+import java.util.List;
public class AMSTableTestBase extends TableServiceTestBase {
@ClassRule public static TestHMS TEST_HMS = new TestHMS();
@@ -79,15 +82,17 @@ public class AMSTableTestBase extends TableServiceTestBase {
public void init() throws IOException, TException {
catalogWarehouse = temp.newFolder().getPath();
catalogMeta = catalogTestHelper.buildCatalogMeta(catalogWarehouse);
- if (catalogTestHelper.isInternalCatalog()
- || TableFormat.MIXED_HIVE.equals(catalogTestHelper.tableFormat())) {
- if (TableFormat.MIXED_ICEBERG.equals(catalogTestHelper.tableFormat())
- || TableFormat.MIXED_HIVE.equals(catalogTestHelper.tableFormat())) {
+ // mixed-hive format only exists in external catalog
+ if (catalogTestHelper.isInternalCatalog()) {
+ if (TableFormat.MIXED_ICEBERG.equals(catalogTestHelper.tableFormat())) {
mixedTables = catalogTestHelper.buildMixedTables(catalogMeta);
tableMeta = buildTableMeta();
}
} else {
externalCatalog = new CommonUnifiedCatalog(() -> catalogMeta,
Maps.newHashMap());
+ if (TableFormat.MIXED_HIVE.equals(catalogTestHelper.tableFormat())) {
+ tableMeta = buildTableMeta();
+ }
}
tableService().createCatalog(catalogMeta);
@@ -143,7 +148,10 @@ public class AMSTableTestBase extends TableServiceTestBase
{
.createDatabase(TableTestHelper.TEST_CATALOG_NAME,
TableTestHelper.TEST_DB_NAME);
}
} else {
- externalCatalog.createDatabase(TableTestHelper.TEST_DB_NAME);
+ List<String> databases = externalCatalog.listDatabases();
+ if (!(databases != null &&
databases.contains(TableTestHelper.TEST_DB_NAME))) {
+ externalCatalog.createDatabase(TableTestHelper.TEST_DB_NAME);
+ }
}
}
@@ -172,6 +180,9 @@ public class AMSTableTestBase extends TableServiceTestBase {
case MIXED_ICEBERG:
createMixedIcebergTable();
break;
+ case MIXED_HIVE:
+ createMixedHiveTable();
+ break;
default:
throw new IllegalStateException("un-support format");
}
@@ -181,6 +192,25 @@ public class AMSTableTestBase extends TableServiceTestBase
{
serverTableIdentifier = tableService().listManagedTables().get(0);
}
+ private void createMixedHiveTable() {
+ // only create mixed hive table here !
+ catalogMeta.putToCatalogProperties(
+ CatalogMetaProperties.TABLE_FORMATS, TableFormat.MIXED_HIVE.name());
+ ArcticHiveCatalog catalog =
+ (ArcticHiveCatalog)
+ CatalogLoader.createCatalog(
+ catalogMeta.getCatalogName(),
+ catalogMeta.getCatalogType(),
+ catalogMeta.getCatalogProperties(),
+ ArcticCatalogUtil.buildMetaStore(catalogMeta));
+ catalog
+ .newTableBuilder(tableTestHelper.id(), tableTestHelper.tableSchema())
+ .withPartitionSpec(tableTestHelper.partitionSpec())
+ .withProperties(tableTestHelper.tableProperties())
+ .withPrimaryKeySpec(tableTestHelper.primaryKeySpec())
+ .create();
+ }
+
private void createMixedIcebergTable() {
ArcticCatalog catalog =
CatalogLoader.createCatalog(
diff --git
a/ams/server/src/test/java/com/netease/arctic/server/table/TestTableRuntimeHandler.java
b/ams/server/src/test/java/com/netease/arctic/server/table/TestTableRuntimeHandler.java
index d33ebde1d..ed678d176 100644
---
a/ams/server/src/test/java/com/netease/arctic/server/table/TestTableRuntimeHandler.java
+++
b/ams/server/src/test/java/com/netease/arctic/server/table/TestTableRuntimeHandler.java
@@ -66,12 +66,15 @@ public class TestTableRuntimeHandler extends
AMSTableTestBase {
}
@Test
- public void testInitialize() {
+ public void testInitialize() throws Exception {
tableService = new DefaultTableService(new Configurations());
TestHandler handler = new TestHandler();
tableService.addHandlerChain(handler);
tableService.initialize();
- createDatabase();
+ if (!(catalogTestHelper().tableFormat().equals(TableFormat.MIXED_HIVE)
+ && TEST_HMS.getHiveClient().getDatabase(TableTestHelper.TEST_DB_NAME)
!= null)) {
+ createDatabase();
+ }
createTable();
ServerTableIdentifier createTableId =
tableService().listManagedTables().get(0);
Assert.assertEquals(1, handler.getAddedTables().size());
diff --git
a/ams/server/src/test/java/com/netease/arctic/server/table/TestTableService.java
b/ams/server/src/test/java/com/netease/arctic/server/table/TestTableService.java
index f5faac959..42504d5ee 100644
---
a/ams/server/src/test/java/com/netease/arctic/server/table/TestTableService.java
+++
b/ams/server/src/test/java/com/netease/arctic/server/table/TestTableService.java
@@ -70,13 +70,17 @@ public class TestTableService extends AMSTableTestBase {
@Test
public void testCreateAndDropTable() {
- tableService().createDatabase(TEST_CATALOG_NAME, TEST_DB_NAME);
+ if (catalogTestHelper().isInternalCatalog()) {
+ tableService().createDatabase(TEST_CATALOG_NAME, TEST_DB_NAME);
+ }
// test create table
createTable();
- Assert.assertEquals(
- tableMeta(),
-
tableService().loadTableMetadata(tableMeta().getTableIdentifier()).buildTableMeta());
+ if (catalogTestHelper().isInternalCatalog()) {
+ Assert.assertEquals(
+ tableMeta(),
+
tableService().loadTableMetadata(tableMeta().getTableIdentifier()).buildTableMeta());
+ }
// test list tables
List<TableIDWithFormat> tableIdentifierList =
@@ -87,13 +91,14 @@ public class TestTableService extends AMSTableTestBase {
tableIdentifierList.get(0).getIdentifier().buildTableIdentifier());
// test list table metadata
- List<TableMetadata> tableMetadataList = tableService().listTableMetas();
- Assert.assertEquals(1, tableMetadataList.size());
- Assert.assertEquals(tableMeta(),
tableMetadataList.get(0).buildTableMeta());
-
- tableMetadataList = tableService().listTableMetas(TEST_CATALOG_NAME,
TEST_DB_NAME);
- Assert.assertEquals(1, tableMetadataList.size());
- Assert.assertEquals(tableMeta(),
tableMetadataList.get(0).buildTableMeta());
+ if (catalogTestHelper().isInternalCatalog()) {
+ List<TableMetadata> tableMetadataList = tableService().listTableMetas();
+ Assert.assertEquals(1, tableMetadataList.size());
+ Assert.assertEquals(tableMeta(),
tableMetadataList.get(0).buildTableMeta());
+ tableMetadataList = tableService().listTableMetas(TEST_CATALOG_NAME,
TEST_DB_NAME);
+ Assert.assertEquals(1, tableMetadataList.size());
+ Assert.assertEquals(tableMeta(),
tableMetadataList.get(0).buildTableMeta());
+ }
// test table exist
Assert.assertTrue(tableService().tableExist(tableMeta().getTableIdentifier()));
@@ -148,7 +153,7 @@ public class TestTableService extends AMSTableTestBase {
ObjectNotExistsException.class,
() ->
tableService().dropTableMetadata(tableMeta().getTableIdentifier(), true));
- tableService().dropDatabase(TEST_CATALOG_NAME, TEST_DB_NAME);
+ dropDatabase();
}
@Test
diff --git
a/core/src/main/java/com/netease/arctic/catalog/BasicArcticCatalog.java
b/core/src/main/java/com/netease/arctic/catalog/BasicArcticCatalog.java
index 767b97814..7360df9a1 100644
--- a/core/src/main/java/com/netease/arctic/catalog/BasicArcticCatalog.java
+++ b/core/src/main/java/com/netease/arctic/catalog/BasicArcticCatalog.java
@@ -326,6 +326,17 @@ public class BasicArcticCatalog implements ArcticCatalog {
return tables.createTableByMeta(tableMeta, schema, primaryKeySpec,
partitionSpec);
}
+ protected void checkAmsTableMetadata() {
+ try {
+ client.getTable(identifier.buildTableIdentifier());
+ throw new org.apache.iceberg.exceptions.AlreadyExistsException("table
already exist");
+ } catch (NoSuchObjectException e) {
+ checkProperties();
+ } catch (TException e) {
+ throw new IllegalStateException("failed when load table", e);
+ }
+ }
+
protected void doCreateCheck() {
if (primaryKeySpec.primaryKeyExisted()) {
primaryKeySpec
@@ -345,14 +356,7 @@ public class BasicArcticCatalog implements ArcticCatalog {
.findFirst()
.orElseThrow(() -> new
NoSuchDatabaseException(identifier.getDatabase()));
- try {
- client.getTable(identifier.buildTableIdentifier());
- throw new org.apache.iceberg.exceptions.AlreadyExistsException("table
already exist");
- } catch (NoSuchObjectException e) {
- checkProperties();
- } catch (TException e) {
- throw new IllegalStateException("failed when load table", e);
- }
+ checkAmsTableMetadata();
}
protected void checkProperties() {
diff --git a/core/src/main/java/com/netease/arctic/catalog/MixedTables.java
b/core/src/main/java/com/netease/arctic/catalog/MixedTables.java
index 93f548fdc..9c2823ed6 100644
--- a/core/src/main/java/com/netease/arctic/catalog/MixedTables.java
+++ b/core/src/main/java/com/netease/arctic/catalog/MixedTables.java
@@ -67,6 +67,10 @@ public class MixedTables {
this.tables = new HadoopTables(tableMetaStore.getConfiguration());
}
+ public Table loadHadoopTableByLocation(String location) {
+ return tableMetaStore.doAs(() -> tables.load(location));
+ }
+
public ArcticTable loadTableByMeta(TableMeta tableMeta) {
if (tableMeta.getKeySpec() != null
&& tableMeta.getKeySpec().getFields() != null
diff --git
a/mixed-format/hive/src/main/java/com/netease/arctic/hive/HMSClient.java
b/mixed-format/hive/src/main/java/com/netease/arctic/hive/HMSClient.java
index feb89a197..074a81e7c 100644
--- a/mixed-format/hive/src/main/java/com/netease/arctic/hive/HMSClient.java
+++ b/mixed-format/hive/src/main/java/com/netease/arctic/hive/HMSClient.java
@@ -86,4 +86,6 @@ public interface HMSClient {
EnvironmentContext environmentContext)
throws TException, InstantiationException, IllegalAccessException,
NoSuchMethodException,
InvocationTargetException, ClassNotFoundException;
+
+ List<Table> getTableObjectsByName(String dbName, List<String> tableNames)
throws TException;
}
diff --git
a/mixed-format/hive/src/main/java/com/netease/arctic/hive/HMSClientImpl.java
b/mixed-format/hive/src/main/java/com/netease/arctic/hive/HMSClientImpl.java
index 657b27b96..37d548aec 100644
--- a/mixed-format/hive/src/main/java/com/netease/arctic/hive/HMSClientImpl.java
+++ b/mixed-format/hive/src/main/java/com/netease/arctic/hive/HMSClientImpl.java
@@ -181,4 +181,10 @@ public class HMSClientImpl implements HMSClient {
.build();
alterPartition.invoke(getClient(), dbName, tblName, newPart,
environmentContext);
}
+
+ @Override
+ public List<Table> getTableObjectsByName(String dbName, List<String>
tableNames)
+ throws TException {
+ return getClient().getTableObjectsByName(dbName, tableNames);
+ }
}
diff --git
a/mixed-format/hive/src/main/java/com/netease/arctic/hive/catalog/ArcticHiveCatalog.java
b/mixed-format/hive/src/main/java/com/netease/arctic/hive/catalog/ArcticHiveCatalog.java
index 2df9ba5b7..4624de98f 100644
---
a/mixed-format/hive/src/main/java/com/netease/arctic/hive/catalog/ArcticHiveCatalog.java
+++
b/mixed-format/hive/src/main/java/com/netease/arctic/hive/catalog/ArcticHiveCatalog.java
@@ -18,21 +18,45 @@
package com.netease.arctic.hive.catalog;
+import static
com.netease.arctic.properties.HiveTableProperties.ARCTIC_TABLE_PRIMARY_KEYS;
+import static
com.netease.arctic.properties.HiveTableProperties.ARCTIC_TABLE_ROOT_LOCATION;
+import static
com.netease.arctic.table.PrimaryKeySpec.PRIMARY_KEY_COLUMN_JOIN_DELIMITER;
+import static
com.netease.arctic.table.TableProperties.LOG_STORE_STORAGE_TYPE_KAFKA;
+import static
com.netease.arctic.table.TableProperties.LOG_STORE_STORAGE_TYPE_PULSAR;
+import static com.netease.arctic.table.TableProperties.LOG_STORE_TYPE;
+
+import com.netease.arctic.AmsClient;
+import com.netease.arctic.NoSuchDatabaseException;
+import com.netease.arctic.PooledAmsClient;
import com.netease.arctic.TableFormat;
import com.netease.arctic.api.TableMeta;
-import com.netease.arctic.catalog.BasicArcticCatalog;
-import com.netease.arctic.catalog.MixedTables;
+import com.netease.arctic.catalog.ArcticCatalog;
import com.netease.arctic.hive.CachedHiveClientPool;
import com.netease.arctic.hive.HMSClient;
import com.netease.arctic.hive.HMSClientPool;
+import com.netease.arctic.hive.utils.CompatibleHivePropertyUtil;
import com.netease.arctic.hive.utils.HiveSchemaUtil;
+import com.netease.arctic.hive.utils.HiveTableUtil;
+import com.netease.arctic.io.ArcticFileIO;
+import com.netease.arctic.io.ArcticFileIOs;
+import com.netease.arctic.op.ArcticHadoopTableOperations;
+import com.netease.arctic.op.CreateTableTransaction;
+import com.netease.arctic.properties.CatalogMetaProperties;
import com.netease.arctic.properties.HiveTableProperties;
+import com.netease.arctic.properties.MetaTableProperties;
+import com.netease.arctic.table.ArcticTable;
import com.netease.arctic.table.PrimaryKeySpec;
import com.netease.arctic.table.TableBuilder;
import com.netease.arctic.table.TableIdentifier;
import com.netease.arctic.table.TableMetaStore;
import com.netease.arctic.table.TableProperties;
+import com.netease.arctic.table.blocker.BasicTableBlockerManager;
+import com.netease.arctic.table.blocker.TableBlockerManager;
+import com.netease.arctic.utils.ArcticCatalogUtil;
+import com.netease.arctic.utils.CompatiblePropertyUtil;
import com.netease.arctic.utils.ConvertStructUtil;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.metastore.api.AlreadyExistsException;
import org.apache.hadoop.hive.metastore.api.Database;
import org.apache.hadoop.hive.metastore.api.NoSuchObjectException;
@@ -41,33 +65,64 @@ import org.apache.iceberg.PartitionField;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.SortOrder;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.TableOperations;
+import org.apache.iceberg.Transaction;
+import org.apache.iceberg.Transactions;
+import org.apache.iceberg.exceptions.NoSuchTableException;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.thrift.TException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
+import java.util.Objects;
+import java.util.stream.Collectors;
/**
* Implementation of {@link com.netease.arctic.catalog.ArcticCatalog} to
support Hive table as base
* store.
*/
-public class ArcticHiveCatalog extends BasicArcticCatalog {
+public class ArcticHiveCatalog implements ArcticCatalog {
private static final Logger LOG =
LoggerFactory.getLogger(ArcticHiveCatalog.class);
+ protected AmsClient client;
private CachedHiveClientPool hiveClientPool;
+ protected String name;
+ protected Map<String, String> catalogProperties;
+ protected MixedHiveTables tables;
+ protected transient TableMetaStore tableMetaStore;
+
+ @Override
+ public String name() {
+ return name;
+ }
@Override
public void initialize(String name, Map<String, String> properties,
TableMetaStore metaStore) {
- super.initialize(name, properties, metaStore);
+ if (properties.get(CatalogMetaProperties.AMS_URI) != null) {
+ this.client = new
PooledAmsClient(properties.get(CatalogMetaProperties.AMS_URI));
+ }
+ this.name = name;
+ this.catalogProperties = properties;
+ this.tableMetaStore = metaStore;
+ this.tables = newMixedHiveTables(properties, metaStore);
this.hiveClientPool = ((MixedHiveTables) tables).getHiveClientPool();
}
- @Override
- protected MixedTables newMixedTables(
+ protected MixedHiveTables getTables() {
+ return tables;
+ }
+
+ protected MixedHiveTables newMixedHiveTables(
Map<String, String> catalogProperties, TableMetaStore metaStore) {
return new MixedHiveTables(catalogProperties, metaStore);
}
@@ -100,10 +155,93 @@ public class ArcticHiveCatalog extends BasicArcticCatalog
{
}
}
- /** HMS is case-insensitive for table name and database */
@Override
+ public TableBlockerManager getTableBlockerManager(TableIdentifier
tableIdentifier) {
+ validate(tableIdentifier);
+ return BasicTableBlockerManager.build(tableIdentifier, client);
+ }
+
+ @Override
+ public Map<String, String> properties() {
+ return catalogProperties;
+ }
+
+ public static void putNotNullProperties(
+ Map<String, String> properties, String key, String value) {
+ if (value != null) {
+ properties.put(key, value);
+ }
+ }
+
+ /** HMS is case-insensitive for table name and database */
protected TableMeta getArcticTableMeta(TableIdentifier identifier) {
- return super.getArcticTableMeta(identifier.toLowCaseIdentifier());
+ org.apache.hadoop.hive.metastore.api.Table hiveTable = null;
+ identifier = identifier.toLowCaseIdentifier();
+ try {
+ hiveTable = HiveTableUtil.loadHmsTable(this.hiveClientPool, identifier);
+ } catch (RuntimeException e) {
+ throw new IllegalStateException(String.format("failed load hive table
%s.", identifier), e);
+ }
+ if (hiveTable == null) {
+ throw new NoSuchTableException("load table failed %s.", identifier);
+ }
+
+ Map<String, String> hiveParameters = hiveTable.getParameters();
+
+ String arcticRootLocation = hiveParameters.get(ARCTIC_TABLE_ROOT_LOCATION);
+ if (arcticRootLocation == null) {
+ // if hive location ends with /hive, then it's a mixed-hive table. we
need to remove /hive to
+ // get root location.
+ // if hive location doesn't end with /hive, then it's a pure-hive table.
we can use the
+ // location as root location.
+ String hiveRootLocation = hiveTable.getSd().getLocation();
+ if (hiveRootLocation.endsWith("/hive")) {
+ arcticRootLocation = hiveRootLocation.substring(0,
hiveRootLocation.length() - 5);
+ } else {
+ arcticRootLocation = hiveRootLocation;
+ }
+ }
+
+ // full path of base, change and root location
+ String baseLocation = arcticRootLocation + "/base";
+ String changeLocation = arcticRootLocation + "/change";
+ // load base table for get arctic table properties
+ Table baseIcebergTable =
getTables().loadHadoopTableByLocation(baseLocation);
+ if (baseIcebergTable == null) {
+ throw new NoSuchTableException("load table failed %s, base table not
found.", identifier);
+ }
+ Map<String, String> properties = baseIcebergTable.properties();
+ // start to construct TableMeta
+ TableMeta tableMeta = new TableMeta();
+ tableMeta.setTableIdentifier(identifier.buildTableIdentifier());
+
+ Map<String, String> locations = new HashMap<>();
+ putNotNullProperties(locations, MetaTableProperties.LOCATION_KEY_TABLE,
arcticRootLocation);
+ putNotNullProperties(locations, MetaTableProperties.LOCATION_KEY_CHANGE,
changeLocation);
+ putNotNullProperties(locations, MetaTableProperties.LOCATION_KEY_BASE,
baseLocation);
+ // set table location
+ tableMeta.setLocations(locations);
+
+ // set table properties
+ Map<String, String> newProperties = new HashMap<>(properties);
+ tableMeta.setProperties(newProperties);
+
+ // set table's primary key when needed
+ if (hiveParameters != null) {
+ String primaryKey = hiveParameters.get(ARCTIC_TABLE_PRIMARY_KEYS);
+ // primary key info come from hive properties
+ if (StringUtils.isNotBlank(primaryKey)) {
+ com.netease.arctic.api.PrimaryKeySpec keySpec = new
com.netease.arctic.api.PrimaryKeySpec();
+ List<String> fields =
+ Arrays.stream(primaryKey.split(PRIMARY_KEY_COLUMN_JOIN_DELIMITER))
+ .collect(Collectors.toList());
+ keySpec.setFields(fields);
+ tableMeta.setKeySpec(keySpec);
+ }
+ }
+ // set table format to mixed-hive format
+ tableMeta.setFormat(TableFormat.MIXED_HIVE.name());
+ return tableMeta;
}
@Override
@@ -130,26 +268,258 @@ public class ArcticHiveCatalog extends
BasicArcticCatalog {
return new MixedHiveTableBuilder(identifier, schema);
}
+ @Override
+ public void renameTable(TableIdentifier from, String newTableName) {
+ throw new UnsupportedOperationException("unsupported rename arctic table
for now.");
+ }
+
public HMSClientPool getHMSClient() {
return hiveClientPool;
}
- class MixedHiveTableBuilder extends ArcticTableBuilder {
+ /**
+ *
+ *
+ * <ul>
+ * <li>1、call getTableObjectsByName to get all Table objects of database
+ * <li>2、filter hive tables whose properties don't have arctic table flag
+ * </ul>
+ *
+ * we don't do cache here because we create/drop table through engine (like
spark) connector, they
+ * have another ArcticHiveCatalog instance。 we can't find a easy way to
update cache.
+ *
+ * @param database
+ * @return
+ */
+ @Override
+ public List<TableIdentifier> listTables(String database) {
+ final List<TableIdentifier> result = new ArrayList<>();
+ try {
+ hiveClientPool.run(
+ client -> {
+ List<String> tableNames = client.getAllTables(database);
+ long start = System.currentTimeMillis();
+ List<org.apache.hadoop.hive.metastore.api.Table> hiveTables =
+ client.getTableObjectsByName(database, tableNames);
+ LOG.info("call getTableObjectsByName cost {} ms",
System.currentTimeMillis() - start);
+ // filter hive tables whose properties don't have arctic table flag
+ if (hiveTables != null && !hiveTables.isEmpty()) {
+ List<TableIdentifier> loadResult =
+ hiveTables.stream()
+ .filter(
+ table ->
+ table.getParameters() != null
+ &&
CompatibleHivePropertyUtil.propertyAsBoolean(
+ table.getParameters(),
+ HiveTableProperties.ARCTIC_TABLE_FLAG,
+ false))
+ .map(table -> TableIdentifier.of(name(), database,
table.getTableName()))
+ .collect(Collectors.toList());
+ if (loadResult != null && !loadResult.isEmpty()) {
+ result.addAll(loadResult);
+ }
+ LOG.debug(
+ "load {} tables from database {} of catalog {}",
+ loadResult == null ? 0 : loadResult.size(),
+ database,
+ name());
+ } else {
+ LOG.debug("load no tables from database {} of catalog {}",
database, name());
+ }
+ return result;
+ });
+ } catch (NoSuchObjectException e) {
+ // pass
+ } catch (TException | InterruptedException e) {
+ throw new RuntimeException("Failed to listTables of database :" +
database, e);
+ }
+ return result;
+ }
+
+ private void validate(TableIdentifier identifier) {
+ if (StringUtils.isNotBlank(identifier.getCatalog())) {
+ identifier.setCatalog(this.name());
+ } else if (!this.name().equals(identifier.getCatalog())) {
+ throw new IllegalArgumentException("catalog name miss match");
+ }
+ }
+
+ @Override
+ public ArcticTable loadTable(TableIdentifier identifier) {
+ validate(identifier);
+ TableMeta meta = getArcticTableMeta(identifier);
+ if (meta.getLocations() == null) {
+ throw new IllegalStateException("load table failed, lack locations
info");
+ }
+ return tables.loadTableByMeta(meta);
+ }
+
+ @Override
+ public boolean dropTable(TableIdentifier identifier, boolean purge) {
+ validate(identifier);
+ TableMeta meta;
+ try {
+ meta = getArcticTableMeta(identifier);
+ } catch (NoSuchTableException e) {
+ return false;
+ }
+
+ doDropTable(meta, purge);
+ return true;
+ }
+
+ protected void doDropTable(TableMeta meta, boolean purge) {
+ tables.dropTableByMeta(meta, purge);
+ }
+
+ class MixedHiveTableBuilder implements TableBuilder {
+ protected TableIdentifier identifier;
+ protected Schema schema;
+ protected PartitionSpec partitionSpec;
+ protected SortOrder sortOrder;
+ protected Map<String, String> properties = new HashMap<>();
+ protected PrimaryKeySpec primaryKeySpec = PrimaryKeySpec.noPrimaryKey();
+ protected String location;
+
+ @Override
+ public ArcticTable create() {
+ ConvertStructUtil.TableMetaBuilder builder = createTableMataBuilder();
+ doCreateCheck();
+ TableMeta meta = builder.build();
+ ArcticTable table = createTableByMeta(meta, schema, primaryKeySpec,
partitionSpec);
+ return table;
+ }
+
+ @Override
+ public Transaction createTransaction() {
+ ArcticFileIO arcticFileIO =
ArcticFileIOs.buildHadoopFileIO(tableMetaStore);
+ ConvertStructUtil.TableMetaBuilder builder = createTableMataBuilder();
+ TableMeta meta = builder.build();
+ String location = getTableLocationForCreate();
+ TableOperations tableOperations =
+ new ArcticHadoopTableOperations(
+ new Path(location), arcticFileIO,
tableMetaStore.getConfiguration());
+ TableMetadata tableMetadata =
+ tableMetadata(schema, partitionSpec, sortOrder, properties,
location);
+ Transaction transaction =
+ Transactions.createTableTransaction(
+ identifier.getTableName(), tableOperations, tableMetadata);
+ return new CreateTableTransaction(
+ transaction,
+ this::create,
+ () -> {
+ doRollbackCreateTable(meta);
+ try {
+ client.removeTable(identifier.buildTableIdentifier(), true);
+ } catch (TException e) {
+ throw new RuntimeException(e);
+ }
+ });
+ }
+
+ protected ArcticTable createTableByMeta(
+ TableMeta tableMeta,
+ Schema schema,
+ PrimaryKeySpec primaryKeySpec,
+ PartitionSpec partitionSpec) {
+ return tables.createTableByMeta(tableMeta, schema, primaryKeySpec,
partitionSpec);
+ }
+
+ protected void checkProperties() {
+ Map<String, String> mergedProperties =
+ ArcticCatalogUtil.mergeCatalogPropertiesToTable(properties,
catalogProperties);
+ boolean enableStream =
+ CompatiblePropertyUtil.propertyAsBoolean(
+ mergedProperties,
+ TableProperties.ENABLE_LOG_STORE,
+ TableProperties.ENABLE_LOG_STORE_DEFAULT);
+ if (enableStream) {
+ Preconditions.checkArgument(
+
mergedProperties.containsKey(TableProperties.LOG_STORE_MESSAGE_TOPIC),
+ "log-store.topic must not be null when log-store.enabled is
true.");
+ Preconditions.checkArgument(
+ mergedProperties.containsKey(TableProperties.LOG_STORE_ADDRESS),
+ "log-store.address must not be null when log-store.enabled is
true.");
+ String logStoreType = mergedProperties.get(LOG_STORE_TYPE);
+ Preconditions.checkArgument(
+ logStoreType == null
+ || logStoreType.equals(LOG_STORE_STORAGE_TYPE_KAFKA)
+ || logStoreType.equals(LOG_STORE_STORAGE_TYPE_PULSAR),
+ String.format(
+ "%s can not be set %s, valid values are: [%s, %s].",
+ LOG_STORE_TYPE,
+ logStoreType,
+ LOG_STORE_STORAGE_TYPE_KAFKA,
+ LOG_STORE_STORAGE_TYPE_PULSAR));
+ properties.putIfAbsent(
+ TableProperties.LOG_STORE_DATA_FORMAT,
TableProperties.LOG_STORE_DATA_FORMAT_DEFAULT);
+ }
+ }
+
+ private String getTableLocationForCreate() {
+ if (StringUtils.isNotBlank(location)) {
+ return location;
+ }
+
+ if (properties.containsKey(TableProperties.LOCATION)) {
+ String tableLocation = properties.get(TableProperties.LOCATION);
+ if (!Objects.equals("/", tableLocation) &&
tableLocation.endsWith("/")) {
+ tableLocation = tableLocation.substring(0, tableLocation.length() -
1);
+ }
+ if (StringUtils.isNotBlank(tableLocation)) {
+ return tableLocation;
+ }
+ }
+
+ String databaseLocation = getDatabaseLocation();
+
+ if (StringUtils.isNotBlank(databaseLocation)) {
+ return databaseLocation + '/' + identifier.getTableName();
+ } else {
+ throw new IllegalStateException(
+ "either `location` in table properties or "
+ + "`warehouse` in catalog properties is specified");
+ }
+ }
+
+ protected TableMetadata tableMetadata(
+ Schema schema,
+ PartitionSpec spec,
+ SortOrder order,
+ Map<String, String> properties,
+ String location) {
+ Preconditions.checkNotNull(schema, "A table schema is required");
+
+ Map<String, String> tableProps = properties == null ? ImmutableMap.of()
: properties;
+ PartitionSpec partitionSpec = spec == null ?
PartitionSpec.unpartitioned() : spec;
+ SortOrder sortOrder = order == null ? SortOrder.unsorted() : order;
+ return TableMetadata.newTableMetadata(schema, partitionSpec, sortOrder,
location, tableProps);
+ }
public MixedHiveTableBuilder(TableIdentifier identifier, Schema schema) {
- super(identifier.toLowCaseIdentifier(),
HiveSchemaUtil.changeFieldNameToLowercase(schema));
+ Preconditions.checkArgument(
+ identifier.getCatalog().equals(name()),
+ "Illegal table id:%s for catalog:%s",
+ identifier.toString(),
+ name());
+ this.identifier = identifier.toLowCaseIdentifier();
+ this.schema = HiveSchemaUtil.changeFieldNameToLowercase(schema);
+ this.partitionSpec = PartitionSpec.unpartitioned();
+ this.sortOrder = SortOrder.unsorted();
}
boolean allowExistedHiveTable = false;
@Override
public TableBuilder withPartitionSpec(PartitionSpec partitionSpec) {
- return
super.withPartitionSpec(IcebergSchemaUtil.copyPartitionSpec(partitionSpec,
schema));
+ this.partitionSpec = IcebergSchemaUtil.copyPartitionSpec(partitionSpec,
schema);
+ return this;
}
@Override
public TableBuilder withSortOrder(SortOrder sortOrder) {
- return
super.withSortOrder(IcebergSchemaUtil.copySortOrderSpec(sortOrder, schema));
+ this.sortOrder = IcebergSchemaUtil.copySortOrderSpec(sortOrder, schema);
+ return this;
}
@Override
@@ -160,18 +530,19 @@ public class ArcticHiveCatalog extends BasicArcticCatalog
{
.forEach(
primaryKeyField ->
builder.addColumn(primaryKeyField.fieldName().toLowerCase(Locale.ROOT)));
- return super.withPrimaryKeySpec(builder.build());
+ this.primaryKeySpec = builder.build();
+ return this;
}
@Override
public TableBuilder withProperty(String key, String value) {
if (key.equals(HiveTableProperties.ALLOW_HIVE_TABLE_EXISTED) &&
value.equals("true")) {
allowExistedHiveTable = true;
- super.withProperty(key, value);
+ this.properties.put(key, value);
} else if (key.equals(TableProperties.TABLE_EVENT_TIME_FIELD)) {
- super.withProperty(key, value.toLowerCase(Locale.ROOT));
+ this.properties.put(key, value.toLowerCase(Locale.ROOT));
} else {
- super.withProperty(key, value);
+ this.properties.put(key, value);
}
return this;
}
@@ -182,17 +553,40 @@ public class ArcticHiveCatalog extends BasicArcticCatalog
{
return this;
}
- @Override
protected void doCreateCheck() {
+ if (primaryKeySpec.primaryKeyExisted()) {
+ primaryKeySpec
+ .fieldNames()
+ .forEach(
+ primaryKey -> {
+ if (schema.findField(primaryKey).isOptional()) {
+ throw new IllegalArgumentException(
+ "please check your schema, the primary key nested
field must"
+ + " be required and field name is "
+ + primaryKey);
+ }
+ });
+ }
+ listDatabases().stream()
+ .filter(d -> d.equals(identifier.getDatabase()))
+ .findFirst()
+ .orElseThrow(() -> new
NoSuchDatabaseException(identifier.getDatabase()));
- super.doCreateCheck();
try {
+ org.apache.hadoop.hive.metastore.api.Table hiveTable =
+ hiveClientPool.run(
+ client -> client.getTable(identifier.getDatabase(),
identifier.getTableName()));
+ if (hiveTable != null) {
+ // do some check for whether the table has been upgraded!!!
+ if (CompatibleHivePropertyUtil.propertyAsBoolean(
+ hiveTable.getParameters(),
HiveTableProperties.ARCTIC_TABLE_FLAG, false)) {
+ throw new IllegalArgumentException(
+ String.format("Table %s has already been upgraded !",
identifier));
+ }
+ }
if (allowExistedHiveTable) {
LOG.info("No need to check hive table exist");
} else {
- org.apache.hadoop.hive.metastore.api.Table hiveTable =
- hiveClientPool.run(
- client -> client.getTable(identifier.getDatabase(),
identifier.getTableName()));
if (hiveTable != null) {
throw new IllegalArgumentException(
"Table is already existed in hive meta store:" + identifier);
@@ -215,9 +609,10 @@ public class ArcticHiveCatalog extends BasicArcticCatalog {
"Partition field should be at last of " + "schema");
}
}
+
+ checkProperties();
}
- @Override
protected String getDatabaseLocation() {
try {
return hiveClientPool.run(
@@ -228,7 +623,6 @@ public class ArcticHiveCatalog extends BasicArcticCatalog {
}
}
- @Override
protected void doRollbackCreateTable(TableMeta meta) {
if (allowExistedHiveTable) {
LOG.info(
@@ -237,13 +631,28 @@ public class ArcticHiveCatalog extends BasicArcticCatalog
{
meta.getTableIdentifier().getTableName());
tables.dropTableByMeta(meta, false);
} else {
- super.doRollbackCreateTable(meta);
+ tables.dropTableByMeta(meta, true);
}
}
- @Override
protected ConvertStructUtil.TableMetaBuilder createTableMataBuilder() {
- ConvertStructUtil.TableMetaBuilder builder =
super.createTableMataBuilder();
+ ConvertStructUtil.TableMetaBuilder builder =
+ ConvertStructUtil.newTableMetaBuilder(this.identifier, this.schema);
+ String tableLocation = getTableLocationForCreate();
+
+ builder
+ .withTableLocation(tableLocation)
+ .withProperties(this.properties)
+ .withPrimaryKeySpec(this.primaryKeySpec);
+
+ if (this.primaryKeySpec.primaryKeyExisted()) {
+ builder =
+ builder
+ .withBaseLocation(tableLocation + "/base")
+ .withChangeLocation(tableLocation + "/change");
+ } else {
+ builder = builder.withBaseLocation(tableLocation + "/base");
+ }
return builder.withFormat(TableFormat.MIXED_HIVE);
}
}
diff --git
a/mixed-format/hive/src/main/java/com/netease/arctic/hive/catalog/MixedHiveTables.java
b/mixed-format/hive/src/main/java/com/netease/arctic/hive/catalog/MixedHiveTables.java
index 5fcd8a8ed..1343fd0af 100644
---
a/mixed-format/hive/src/main/java/com/netease/arctic/hive/catalog/MixedHiveTables.java
+++
b/mixed-format/hive/src/main/java/com/netease/arctic/hive/catalog/MixedHiveTables.java
@@ -19,7 +19,6 @@
package com.netease.arctic.hive.catalog;
import com.netease.arctic.api.TableMeta;
-import com.netease.arctic.catalog.MixedTables;
import com.netease.arctic.hive.CachedHiveClientPool;
import com.netease.arctic.hive.table.KeyedHiveTable;
import com.netease.arctic.hive.table.UnkeyedHiveTable;
@@ -28,8 +27,10 @@ import com.netease.arctic.hive.utils.HiveTableUtil;
import com.netease.arctic.io.ArcticFileIO;
import com.netease.arctic.io.ArcticFileIOs;
import com.netease.arctic.io.ArcticHadoopFileIO;
+import com.netease.arctic.io.TableTrashManagers;
import com.netease.arctic.properties.HiveTableProperties;
import com.netease.arctic.properties.MetaTableProperties;
+import com.netease.arctic.table.ArcticTable;
import com.netease.arctic.table.ChangeTable;
import com.netease.arctic.table.KeyedTable;
import com.netease.arctic.table.PrimaryKeySpec;
@@ -37,36 +38,66 @@ import com.netease.arctic.table.TableIdentifier;
import com.netease.arctic.table.TableMetaStore;
import com.netease.arctic.table.TableProperties;
import com.netease.arctic.utils.ArcticCatalogUtil;
+import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.hive.metastore.TableType;
import org.apache.iceberg.FileFormat;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Table;
+import org.apache.iceberg.Tables;
import org.apache.iceberg.exceptions.NoSuchTableException;
+import org.apache.iceberg.hadoop.HadoopTables;
import org.apache.iceberg.mapping.MappingUtil;
import org.apache.iceberg.mapping.NameMappingParser;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
import org.apache.iceberg.util.PropertyUtil;
import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.util.HashMap;
import java.util.Locale;
import java.util.Map;
-public class MixedHiveTables extends MixedTables {
+public class MixedHiveTables {
+ private static final Logger LOG =
LoggerFactory.getLogger(MixedHiveTables.class);
+
+ protected Tables tables;
+ protected TableMetaStore tableMetaStore;
+ protected Map<String, String> catalogProperties;
private volatile CachedHiveClientPool hiveClientPool;
public MixedHiveTables(Map<String, String> catalogProperties, TableMetaStore
metaStore) {
- super(catalogProperties, metaStore);
+ this.tableMetaStore = metaStore;
+ this.catalogProperties = catalogProperties;
+ this.tables = new HadoopTables(tableMetaStore.getConfiguration());
this.hiveClientPool = new CachedHiveClientPool(getTableMetaStore(),
catalogProperties);
}
+ protected TableMetaStore getTableMetaStore() {
+ return tableMetaStore;
+ }
+
public CachedHiveClientPool getHiveClientPool() {
return hiveClientPool;
}
- @Override
+ public Table loadHadoopTableByLocation(String location) {
+ return tableMetaStore.doAs(() -> tables.load(location));
+ }
+
+ public ArcticTable loadTableByMeta(TableMeta tableMeta) {
+ if (tableMeta.getKeySpec() != null
+ && tableMeta.getKeySpec().getFields() != null
+ && tableMeta.getKeySpec().getFields().size() > 0) {
+ return loadKeyedTable(tableMeta);
+ } else {
+ return loadUnKeyedTable(tableMeta);
+ }
+ }
+
protected KeyedHiveTable loadKeyedTable(TableMeta tableMeta) {
TableIdentifier tableIdentifier =
TableIdentifier.of(tableMeta.getTableIdentifier());
String tableLocation = checkLocation(tableMeta,
MetaTableProperties.LOCATION_KEY_TABLE);
@@ -120,7 +151,6 @@ public class MixedHiveTables extends MixedTables {
}
}
- @Override
protected UnkeyedHiveTable loadUnKeyedTable(TableMeta tableMeta) {
TableIdentifier tableIdentifier =
TableIdentifier.of(tableMeta.getTableIdentifier());
String baseLocation = checkLocation(tableMeta,
MetaTableProperties.LOCATION_KEY_BASE);
@@ -144,7 +174,6 @@ public class MixedHiveTables extends MixedTables {
catalogProperties);
}
- @Override
protected KeyedTable createKeyedTable(
TableMeta tableMeta,
Schema schema,
@@ -262,7 +291,6 @@ public class MixedHiveTables extends MixedTables {
tableMeta, tableLocation, primaryKeySpec, hiveClientPool, baseTable,
changeTable);
}
- @Override
protected UnkeyedHiveTable createUnKeyedTable(
TableMeta tableMeta,
Schema schema,
@@ -346,9 +374,68 @@ public class MixedHiveTables extends MixedTables {
catalogProperties);
}
- @Override
+ public void dropInternalTableByMeta(TableMeta tableMeta, boolean purge) {
+ try {
+ ArcticFileIO fileIO = ArcticFileIOs.buildHadoopFileIO(tableMetaStore);
+ Map<String, String> tableProperties = Maps.newHashMap();
+ try {
+ ArcticTable arcticTable = loadTableByMeta(tableMeta);
+ tableProperties.putAll(arcticTable.properties());
+ } catch (Exception loadException) {
+ LOG.warn("load table failed when dropping table", loadException);
+ }
+
+ // If purge is true, all manifest/data files must be located under the
table directory.
+ if (!purge) {
+ String baseLocation =
tableMeta.getLocations().get(MetaTableProperties.LOCATION_KEY_BASE);
+ String changeLocation =
+
tableMeta.getLocations().get(MetaTableProperties.LOCATION_KEY_CHANGE);
+ try {
+ if (StringUtils.isNotBlank(baseLocation)) {
+ dropInternalTable(tableMetaStore, baseLocation, false);
+ }
+ if (StringUtils.isNotBlank(changeLocation)) {
+ dropInternalTable(tableMetaStore, changeLocation, false);
+ }
+ } catch (Exception e) {
+ LOG.warn("drop base/change iceberg table fail ", e);
+ }
+ } else {
+ String tableLocation =
tableMeta.getLocations().get(MetaTableProperties.LOCATION_KEY_TABLE);
+ if (fileIO.exists(tableLocation)) {
+ LOG.info("try to delete table directory location is " +
tableLocation);
+ fileIO.asPrefixFileIO().deletePrefix(tableLocation);
+ }
+ }
+
+ // delete custom trash location
+ String customTrashLocation =
+
tableProperties.get(TableProperties.TABLE_TRASH_CUSTOM_ROOT_LOCATION);
+ if (customTrashLocation != null) {
+ TableIdentifier tableId =
TableIdentifier.of(tableMeta.getTableIdentifier());
+ String trashParentLocation =
+ TableTrashManagers.getTrashParentLocation(tableId,
customTrashLocation);
+ if (fileIO.exists(trashParentLocation)) {
+ fileIO.asPrefixFileIO().deletePrefix(trashParentLocation);
+ }
+ }
+ } catch (Exception e) {
+ LOG.warn("drop table directory fail ", e);
+ }
+ }
+
+ private void dropInternalTable(
+ TableMetaStore tableMetaStore, String internalTableLocation, boolean
purge) {
+ final HadoopTables internalTables = new
HadoopTables(tableMetaStore.getConfiguration());
+ tableMetaStore.doAs(
+ () -> {
+ internalTables.dropTable(internalTableLocation, purge);
+ return null;
+ });
+ }
+
public void dropTableByMeta(TableMeta tableMeta, boolean purge) {
- super.dropTableByMeta(tableMeta, purge);
+ dropInternalTableByMeta(tableMeta, purge);
if (!HiveTableUtil.checkExist(
hiveClientPool, TableIdentifier.of(tableMeta.getTableIdentifier()))) {
// If hive table does not exist, we will not try to drop hive table
@@ -397,7 +484,12 @@ public class MixedHiveTables extends MixedTables {
}
protected void fillTableProperties(TableMeta meta) {
- super.fillTableProperties(meta);
+ meta.putToProperties(
+ TableProperties.TABLE_CREATE_TIME,
String.valueOf(System.currentTimeMillis()));
+ meta.putToProperties(org.apache.iceberg.TableProperties.FORMAT_VERSION,
"2");
+ meta.putToProperties(
+
org.apache.iceberg.TableProperties.METADATA_DELETE_AFTER_COMMIT_ENABLED,
"true");
+ meta.putToProperties("flink.max-continuous-empty-commits",
String.valueOf(Integer.MAX_VALUE));
String tableLocation = checkLocation(meta,
MetaTableProperties.LOCATION_KEY_TABLE);
String hiveLocation = HiveTableUtil.hiveRootLocation(tableLocation);
meta.putToProperties(HiveTableProperties.BASE_HIVE_LOCATION_ROOT,
hiveLocation);
@@ -455,4 +547,34 @@ public class MixedHiveTables extends MixedTables {
tableMeta.getProperties().remove(HiveTableProperties.ALLOW_HIVE_TABLE_EXISTED);
return Boolean.parseBoolean(allowStringValue);
}
+
+ public ArcticTable createTableByMeta(
+ TableMeta tableMeta,
+ Schema schema,
+ PrimaryKeySpec primaryKeySpec,
+ PartitionSpec partitionSpec) {
+ if (primaryKeySpec.primaryKeyExisted()) {
+ return createKeyedTable(tableMeta, schema, primaryKeySpec,
partitionSpec);
+ } else {
+ return createUnKeyedTable(tableMeta, schema, primaryKeySpec,
partitionSpec);
+ }
+ }
+
+ protected String checkLocation(TableMeta meta, String locationKey) {
+ String location = meta.getLocations().get(locationKey);
+ Preconditions.checkArgument(StringUtils.isNotBlank(location), "table
location can't found");
+ return location;
+ }
+
+ protected PrimaryKeySpec buildPrimaryKeySpec(Schema schema, TableMeta
tableMeta) {
+ PrimaryKeySpec.Builder builder = PrimaryKeySpec.builderFor(schema);
+ if (tableMeta.getKeySpec() != null
+ && tableMeta.getKeySpec().getFields() != null
+ && !tableMeta.getKeySpec().getFields().isEmpty()) {
+ for (String field : tableMeta.getKeySpec().getFields()) {
+ builder.addColumn(field);
+ }
+ }
+ return builder.build();
+ }
}
diff --git
a/mixed-format/hive/src/test/java/com/netease/arctic/hive/catalog/HiveCatalogTestHelper.java
b/mixed-format/hive/src/test/java/com/netease/arctic/hive/catalog/HiveCatalogTestHelper.java
index 6e4358c17..526a14440 100644
---
a/mixed-format/hive/src/test/java/com/netease/arctic/hive/catalog/HiveCatalogTestHelper.java
+++
b/mixed-format/hive/src/test/java/com/netease/arctic/hive/catalog/HiveCatalogTestHelper.java
@@ -73,6 +73,9 @@ public class HiveCatalogTestHelper implements
CatalogTestHelper {
@Override
public UnifiedCatalog buildUnifiedCatalog(CatalogMeta catalogMeta) {
+ catalogMeta
+ .getCatalogProperties()
+ .put(CatalogMetaProperties.TABLE_FORMATS,
TableFormat.MIXED_HIVE.name());
return new CommonUnifiedCatalog(() -> catalogMeta, Maps.newHashMap());
}
@@ -90,7 +93,7 @@ public class HiveCatalogTestHelper implements
CatalogTestHelper {
throw new UnsupportedOperationException(
"Cannot build mixed-tables for table format:" + tableFormat);
}
- return new MixedHiveTables(
+ return new MixedTables(
catalogMeta.getCatalogProperties(),
ArcticCatalogUtil.buildMetaStore(catalogMeta));
}
diff --git
a/mixed-format/spark/v3.2/spark/src/main/scala/com/netease/arctic/spark/ArcticSparkExtensions.scala
b/mixed-format/spark/v3.2/spark/src/main/scala/com/netease/arctic/spark/ArcticSparkExtensions.scala
index 6356ef130..3f41ad620 100644
---
a/mixed-format/spark/v3.2/spark/src/main/scala/com/netease/arctic/spark/ArcticSparkExtensions.scala
+++
b/mixed-format/spark/v3.2/spark/src/main/scala/com/netease/arctic/spark/ArcticSparkExtensions.scala
@@ -25,6 +25,7 @@ import com.netease.arctic.spark.sql.execution
import org.apache.spark.sql.SparkSessionExtensions
import
org.apache.spark.sql.catalyst.analysis.{AlignedRowLevelIcebergCommandCheck,
AlignRowLevelCommandAssignments, CheckMergeIntoTableConditions,
MergeIntoIcebergTableResolutionCheck, ProcedureArgumentCoercion,
ResolveMergeIntoTableReferences, ResolveProcedures, RewriteDeleteFromTable,
RewriteMergeIntoTable, RewriteUpdateTable}
import org.apache.spark.sql.catalyst.optimizer._
+import
org.apache.spark.sql.catalyst.parser.extensions.IcebergSparkSqlExtensionsParser
import
org.apache.spark.sql.execution.datasources.v2.{ExtendedDataSourceV2Strategy,
ExtendedV2Writes, OptimizeMetadataOnlyDeleteFromTable,
ReplaceRewrittenRowLevelCommand, RowLevelCommandScanRelationPushDown}
import
org.apache.spark.sql.execution.dynamicpruning.RowLevelCommandDynamicPruning
@@ -74,6 +75,12 @@ class ArcticSparkExtensions extends (SparkSessionExtensions
=> Unit) {
// arctic strategy rules
extensions.injectPlannerStrategy { spark =>
execution.ExtendedArcticStrategy(spark) }
+
+ // iceberg sql parser extensions
+ extensions.injectParser { case (_, parser) => new
IcebergSparkSqlExtensionsParser(parser) }
+
+ // iceberg procedure analyzer extensions
+ extensions.injectResolutionRule { spark => ResolveProcedures(spark) }
}
}
diff --git
a/mixed-format/spark/v3.3/spark/src/main/scala/com/netease/arctic/spark/ArcticSparkExtensions.scala
b/mixed-format/spark/v3.3/spark/src/main/scala/com/netease/arctic/spark/ArcticSparkExtensions.scala
index 95864babb..6a438f466 100644
---
a/mixed-format/spark/v3.3/spark/src/main/scala/com/netease/arctic/spark/ArcticSparkExtensions.scala
+++
b/mixed-format/spark/v3.3/spark/src/main/scala/com/netease/arctic/spark/ArcticSparkExtensions.scala
@@ -26,6 +26,7 @@ import com.netease.arctic.spark.sql.execution
import org.apache.spark.sql.SparkSessionExtensions
import
org.apache.spark.sql.catalyst.analysis.{AlignedRowLevelIcebergCommandCheck,
AlignRowLevelCommandAssignments, CheckMergeIntoTableConditions,
MergeIntoIcebergTableResolutionCheck, ProcedureArgumentCoercion,
ResolveMergeIntoTableReferences, ResolveProcedures,
RewriteDeleteFromIcebergTable, RewriteMergeIntoTable, RewriteUpdateTable}
import org.apache.spark.sql.catalyst.optimizer._
+import
org.apache.spark.sql.catalyst.parser.extensions.IcebergSparkSqlExtensionsParser
import
org.apache.spark.sql.execution.datasources.v2.{ExtendedDataSourceV2Strategy,
ExtendedV2Writes, OptimizeMetadataOnlyDeleteFromIcebergTable,
ReplaceRewrittenRowLevelCommand, RowLevelCommandScanRelationPushDown}
import
org.apache.spark.sql.execution.dynamicpruning.RowLevelCommandDynamicPruning
@@ -79,6 +80,12 @@ class ArcticSparkExtensions extends (SparkSessionExtensions
=> Unit) {
// arctic strategy rules
extensions.injectPlannerStrategy { spark =>
execution.ExtendedArcticStrategy(spark) }
+
+ // iceberg sql parser extensions
+ extensions.injectParser { case (_, parser) => new
IcebergSparkSqlExtensionsParser(parser) }
+
+ // iceberg procedure analyzer extensions
+ extensions.injectResolutionRule { spark => ResolveProcedures(spark) }
}
}
