the-other-tim-brown commented on code in PR #648:
URL: https://github.com/apache/incubator-xtable/pull/648#discussion_r1957394332
##########
xtable-utilities/src/test/resources/catalogConfig.yaml:
##########
@@ -45,6 +45,7 @@ datasets:
- sourceCatalogTableIdentifier:
tableIdentifier:
hierarchicalId: "source-database-1.source-1"
+ partitionSpec: "cs_sold_date_sk:VALUE"
Review Comment:
If the source is not hudi, then this shouldn't be required right?
##########
xtable-hive-metastore/src/test/java/org/apache/xtable/hms/TestHMSCatalogSyncClient.java:
##########
@@ -72,7 +73,8 @@ private HMSCatalogSyncClient createHMSCatalogSyncClient() {
mockHMSCatalogConfig,
testConfiguration,
mockMetaStoreClient,
- mockTableBuilder);
+ mockTableBuilder,
+ Optional.empty());
Review Comment:
Can we have some test where the option is non-empty so we can assert the
PartitionSyncTool is called properly?
##########
xtable-hive-metastore/src/test/java/org/apache/xtable/hms/table/TestHudiHMSCatalogTableBuilder.java:
##########
@@ -0,0 +1,143 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.xtable.hms.table;
+
+import static
org.apache.xtable.hms.table.HudiHMSCatalogTableBuilder.HUDI_METADATA_CONFIG;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.when;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.MockedStatic;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import org.apache.hudi.common.model.HoodieFileFormat;
+import org.apache.hudi.common.table.HoodieTableConfig;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+
+import org.apache.xtable.hms.HMSCatalogSyncTestBase;
+import org.apache.xtable.hms.HMSSchemaExtractor;
+import org.apache.xtable.hudi.HudiTableManager;
+import org.apache.xtable.hudi.catalog.HudiCatalogTableUtils;
+
+@ExtendWith(MockitoExtension.class)
+public class TestHudiHMSCatalogTableBuilder extends HMSCatalogSyncTestBase {
+
+ @Mock private HoodieTableMetaClient mockMetaClient;
+ @Mock private HudiTableManager mockHudiTableManager;
+ @Mock private HoodieTableConfig mockTableConfig;
+
+ private HudiHMSCatalogTableBuilder mockHudiHMSCatalogTableBuilder;
Review Comment:
This is a confusing name since this turns out to be an actual instance
instead of a mock.
##########
xtable-hive-metastore/src/main/java/org/apache/xtable/hms/HMSCatalogPartitionSyncOperations.java:
##########
@@ -0,0 +1,222 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.xtable.hms;
+
+import static
org.apache.xtable.catalog.CatalogUtils.toHierarchicalTableIdentifier;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import lombok.extern.log4j.Log4j2;
+
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.thrift.TException;
+
+import org.apache.hudi.common.util.CollectionUtils;
+import org.apache.hudi.exception.TableNotFoundException;
+
+import org.apache.xtable.catalog.CatalogPartition;
+import org.apache.xtable.catalog.CatalogPartitionSyncOperations;
+import org.apache.xtable.exception.CatalogSyncException;
+import org.apache.xtable.model.catalog.CatalogTableIdentifier;
+import org.apache.xtable.model.catalog.HierarchicalTableIdentifier;
+
+@Log4j2
+public class HMSCatalogPartitionSyncOperations implements
CatalogPartitionSyncOperations {
+
+ private final IMetaStoreClient metaStoreClient;
+ private final HMSCatalogConfig catalogConfig;
+
+ public HMSCatalogPartitionSyncOperations(
+ IMetaStoreClient metaStoreClient, HMSCatalogConfig hmsCatalogConfig) {
+ this.metaStoreClient = metaStoreClient;
+ this.catalogConfig = hmsCatalogConfig;
+ }
+
+ @Override
+ public List<CatalogPartition> getAllPartitions(CatalogTableIdentifier
catalogTableIdentifier) {
+ HierarchicalTableIdentifier tableIdentifier =
+ toHierarchicalTableIdentifier(catalogTableIdentifier);
+ try {
+ return metaStoreClient
+ .listPartitions(
+ tableIdentifier.getDatabaseName(),
tableIdentifier.getTableName(), (short) -1)
+ .stream()
+ .map(p -> new CatalogPartition(p.getValues(),
p.getSd().getLocation()))
+ .collect(Collectors.toList());
+ } catch (TException e) {
+ throw new CatalogSyncException(
+ "Failed to get all partitions for table " + tableIdentifier, e);
+ }
+ }
+
+ @Override
+ public void addPartitionsToTable(
+ CatalogTableIdentifier catalogTableIdentifier, List<CatalogPartition>
partitionsToAdd) {
+ HierarchicalTableIdentifier tableIdentifier =
+ toHierarchicalTableIdentifier(catalogTableIdentifier);
+ if (partitionsToAdd.isEmpty()) {
+ log.info("No partitions to add for " + tableIdentifier);
Review Comment:
Update all the logging to use `{}` instead of direct string concatenation
##########
xtable-hive-metastore/src/test/java/org/apache/xtable/hms/TestHMSCatalogPartitionSyncOperations.java:
##########
@@ -0,0 +1,490 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.xtable.hms;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import lombok.SneakyThrows;
+
+import org.apache.hadoop.hive.metastore.api.SerDeInfo;
+import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.thrift.TException;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import org.apache.xtable.catalog.CatalogPartition;
+import org.apache.xtable.catalog.CatalogPartitionSyncOperations;
+import org.apache.xtable.exception.CatalogSyncException;
+
+@ExtendWith(MockitoExtension.class)
+public class TestHMSCatalogPartitionSyncOperations extends
HMSCatalogSyncTestBase {
+
+ private CatalogPartitionSyncOperations mockHMSPartitionSyncOperations;
+
+ void setupCommonMocks() {
+ mockHMSPartitionSyncOperations =
Review Comment:
The naming is a bit confusing here, this is not a mock of the
`HMSCatalogPartitionSyncOperations`
##########
xtable-hive-metastore/src/main/java/org/apache/xtable/hms/HMSCatalogPartitionSyncOperations.java:
##########
@@ -0,0 +1,222 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.xtable.hms;
+
+import static
org.apache.xtable.catalog.CatalogUtils.toHierarchicalTableIdentifier;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import lombok.extern.log4j.Log4j2;
+
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.thrift.TException;
+
+import org.apache.hudi.common.util.CollectionUtils;
+import org.apache.hudi.exception.TableNotFoundException;
+
+import org.apache.xtable.catalog.CatalogPartition;
+import org.apache.xtable.catalog.CatalogPartitionSyncOperations;
+import org.apache.xtable.exception.CatalogSyncException;
+import org.apache.xtable.model.catalog.CatalogTableIdentifier;
+import org.apache.xtable.model.catalog.HierarchicalTableIdentifier;
+
+@Log4j2
+public class HMSCatalogPartitionSyncOperations implements
CatalogPartitionSyncOperations {
+
+ private final IMetaStoreClient metaStoreClient;
+ private final HMSCatalogConfig catalogConfig;
+
+ public HMSCatalogPartitionSyncOperations(
+ IMetaStoreClient metaStoreClient, HMSCatalogConfig hmsCatalogConfig) {
+ this.metaStoreClient = metaStoreClient;
+ this.catalogConfig = hmsCatalogConfig;
+ }
+
+ @Override
+ public List<CatalogPartition> getAllPartitions(CatalogTableIdentifier
catalogTableIdentifier) {
+ HierarchicalTableIdentifier tableIdentifier =
+ toHierarchicalTableIdentifier(catalogTableIdentifier);
+ try {
+ return metaStoreClient
+ .listPartitions(
+ tableIdentifier.getDatabaseName(),
tableIdentifier.getTableName(), (short) -1)
+ .stream()
+ .map(p -> new CatalogPartition(p.getValues(),
p.getSd().getLocation()))
+ .collect(Collectors.toList());
+ } catch (TException e) {
+ throw new CatalogSyncException(
+ "Failed to get all partitions for table " + tableIdentifier, e);
+ }
+ }
+
+ @Override
+ public void addPartitionsToTable(
+ CatalogTableIdentifier catalogTableIdentifier, List<CatalogPartition>
partitionsToAdd) {
+ HierarchicalTableIdentifier tableIdentifier =
+ toHierarchicalTableIdentifier(catalogTableIdentifier);
+ if (partitionsToAdd.isEmpty()) {
+ log.info("No partitions to add for " + tableIdentifier);
+ return;
+ }
+ log.info("Adding partitions " + partitionsToAdd.size() + " to table " +
tableIdentifier);
+ try {
+ StorageDescriptor sd =
+ metaStoreClient
+ .getTable(tableIdentifier.getDatabaseName(),
tableIdentifier.getTableName())
+ .getSd();
+ for (List<CatalogPartition> batch :
+ CollectionUtils.batches(partitionsToAdd,
catalogConfig.getMaxPartitionsPerRequest())) {
+ List<org.apache.hadoop.hive.metastore.api.Partition> partitionList =
new ArrayList<>();
+ batch.forEach(
+ partition -> {
+ StorageDescriptor partitionSd = new StorageDescriptor();
+ partitionSd.setCols(sd.getCols());
+ partitionSd.setInputFormat(sd.getInputFormat());
+ partitionSd.setOutputFormat(sd.getOutputFormat());
+ partitionSd.setSerdeInfo(sd.getSerdeInfo());
+
+ partitionSd.setLocation(partition.getStorageLocation());
Review Comment:
There is similar code on line 134 below. Can we move some of this into a
common method to reduce repetition?
##########
xtable-hive-metastore/src/main/java/org/apache/xtable/hms/table/HudiHMSCatalogTableBuilder.java:
##########
@@ -0,0 +1,207 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.xtable.hms.table;
+
+import static
org.apache.xtable.catalog.CatalogUtils.toHierarchicalTableIdentifier;
+
+import java.io.IOException;
+import java.time.ZonedDateTime;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+import lombok.extern.log4j.Log4j2;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.metastore.api.FieldSchema;
+import org.apache.hadoop.hive.metastore.api.SerDeInfo;
+import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.security.UserGroupInformation;
+
+import org.apache.hudi.common.model.HoodieFileFormat;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.util.ConfigUtils;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import org.apache.xtable.catalog.CatalogTableBuilder;
+import org.apache.xtable.exception.CatalogSyncException;
+import org.apache.xtable.hms.HMSCatalogConfig;
+import org.apache.xtable.hms.HMSSchemaExtractor;
+import org.apache.xtable.hudi.HudiTableManager;
+import org.apache.xtable.hudi.catalog.HudiCatalogTableUtils;
+import org.apache.xtable.hudi.catalog.HudiInputFormatUtils;
+import org.apache.xtable.model.InternalTable;
+import org.apache.xtable.model.catalog.CatalogTableIdentifier;
+import org.apache.xtable.model.catalog.HierarchicalTableIdentifier;
+import org.apache.xtable.model.schema.InternalPartitionField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.storage.TableFormat;
+
+@Log4j2
+public class HudiHMSCatalogTableBuilder implements CatalogTableBuilder<Table,
Table> {
+
+ private final HudiTableManager hudiTableManager;
+ private final HMSSchemaExtractor schemaExtractor;
+ private final HMSCatalogConfig hmsCatalogConfig;
+
+ private HoodieTableMetaClient metaClient;
+
+ protected static final String HUDI_METADATA_CONFIG =
"hudi.metadata-listing-enabled";
+
+ public HudiHMSCatalogTableBuilder(
+ HMSCatalogConfig hmsCatalogConfig, Configuration configuration) {
+ this.hudiTableManager = HudiTableManager.of(configuration);
+ this.schemaExtractor = HMSSchemaExtractor.getInstance();
+ this.hmsCatalogConfig = hmsCatalogConfig;
+ }
+
+ @VisibleForTesting
+ HudiHMSCatalogTableBuilder(
+ HMSCatalogConfig hmsCatalogConfig,
+ HMSSchemaExtractor schemaExtractor,
+ HudiTableManager hudiTableManager,
+ HoodieTableMetaClient metaClient) {
+ this.hudiTableManager = hudiTableManager;
+ this.schemaExtractor = schemaExtractor;
+ this.metaClient = metaClient;
+ this.hmsCatalogConfig = hmsCatalogConfig;
+ }
+
+ HoodieTableMetaClient getMetaClient(String basePath) {
+ if (metaClient == null) {
+ Optional<HoodieTableMetaClient> metaClientOpt =
+ hudiTableManager.loadTableMetaClientIfExists(basePath);
+
+ if (!metaClientOpt.isPresent()) {
+ throw new CatalogSyncException(
+ "failed to get meta client since table is not present in the base
path " + basePath);
+ }
+
+ metaClient = metaClientOpt.get();
+ }
+ return metaClient;
+ }
+
+ @Override
+ public Table getCreateTableRequest(
+ InternalTable table, CatalogTableIdentifier catalogTableIdentifier) {
+ HierarchicalTableIdentifier tableIdentifier =
+ toHierarchicalTableIdentifier(catalogTableIdentifier);
+ Table newTb = new Table();
+ newTb.setDbName(tableIdentifier.getDatabaseName());
+ newTb.setTableName(tableIdentifier.getTableName());
+ try {
+ newTb.setOwner(UserGroupInformation.getCurrentUser().getShortUserName());
+ } catch (IOException e) {
+ throw new CatalogSyncException(
+ "Failed to set owner for hms table: " +
tableIdentifier.getTableName(), e);
+ }
+
+ newTb.setCreateTime((int) ZonedDateTime.now().toEpochSecond());
+ List<String> partitionFields =
+ table.getPartitioningFields().stream()
+ .map(field -> field.getSourceField().getName())
+ .collect(Collectors.toList());
+ Map<String, String> tableProperties =
+ getTableProperties(partitionFields, table.getReadSchema());
+ newTb.setParameters(tableProperties);
+ newTb.setSd(getStorageDescriptor(table));
+ newTb.setPartitionKeys(getSchemaPartitionKeys(table));
+ return newTb;
+ }
+
+ @Override
+ public Table getUpdateTableRequest(
+ InternalTable table, Table hmsTable, CatalogTableIdentifier
tableIdentifier) {
+ Map<String, String> parameters = hmsTable.getParameters();
+ List<String> partitionFields =
+ table.getPartitioningFields().stream()
+ .map(field -> field.getSourceField().getName())
+ .collect(Collectors.toList());
+ Map<String, String> tableParameters = hmsTable.getParameters();
+ tableParameters.putAll(getTableProperties(partitionFields,
table.getReadSchema()));
+ hmsTable.setParameters(tableParameters);
+ hmsTable.setSd(getStorageDescriptor(table));
+
+ hmsTable.setParameters(parameters);
+ hmsTable.getSd().setCols(schemaExtractor.toColumns(TableFormat.HUDI,
table.getReadSchema()));
+ return hmsTable;
+ }
+
+ private Map<String, String> getTableProperties(
+ List<String> partitionFields, InternalSchema schema) {
+ Map<String, String> tableProperties = new HashMap<>();
+ tableProperties.put(HUDI_METADATA_CONFIG, "true");
+ Map<String, String> sparkTableProperties =
+ HudiCatalogTableUtils.getSparkTableProperties(
+ partitionFields, "", hmsCatalogConfig.getSchemaLengthThreshold(),
schema);
+ tableProperties.putAll(sparkTableProperties);
+ return tableProperties;
+ }
+
+ @VisibleForTesting
+ StorageDescriptor getStorageDescriptor(InternalTable table) {
+ final StorageDescriptor storageDescriptor = new StorageDescriptor();
+ storageDescriptor.setCols(schemaExtractor.toColumns(TableFormat.HUDI,
table.getReadSchema()));
+ storageDescriptor.setLocation(table.getBasePath());
+ HoodieFileFormat fileFormat =
+
getMetaClient(table.getBasePath()).getTableConfig().getBaseFileFormat();
+ String inputFormatClassName =
HudiInputFormatUtils.getInputFormatClassName(fileFormat, false);
+ String outputFormatClassName =
HudiInputFormatUtils.getOutputFormatClassName(fileFormat);
+ String serdeClassName = HudiInputFormatUtils.getSerDeClassName(fileFormat);
+ storageDescriptor.setInputFormat(inputFormatClassName);
+ storageDescriptor.setOutputFormat(outputFormatClassName);
+ Map<String, String> serdeProperties = getSerdeProperties(false,
table.getBasePath());
+ SerDeInfo serDeInfo = new SerDeInfo();
+ serDeInfo.setSerializationLib(serdeClassName);
+ serDeInfo.setParameters(serdeProperties);
+ storageDescriptor.setSerdeInfo(serDeInfo);
+ return storageDescriptor;
+ }
+
+ private static Map<String, String> getSerdeProperties(boolean
readAsOptimized, String basePath) {
+ Map<String, String> serdeProperties = new HashMap<>();
+ serdeProperties.put(ConfigUtils.TABLE_SERDE_PATH, basePath);
+ serdeProperties.put(ConfigUtils.IS_QUERY_AS_RO_TABLE,
String.valueOf(readAsOptimized));
Review Comment:
This is always false currently, do we need to pass in the `readAsOptimized`
argument?
##########
xtable-hive-metastore/src/main/java/org/apache/xtable/hms/HMSCatalogPartitionSyncOperations.java:
##########
@@ -0,0 +1,222 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.xtable.hms;
+
+import static
org.apache.xtable.catalog.CatalogUtils.toHierarchicalTableIdentifier;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import lombok.extern.log4j.Log4j2;
+
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.thrift.TException;
+
+import org.apache.hudi.common.util.CollectionUtils;
+import org.apache.hudi.exception.TableNotFoundException;
+
+import org.apache.xtable.catalog.CatalogPartition;
+import org.apache.xtable.catalog.CatalogPartitionSyncOperations;
+import org.apache.xtable.exception.CatalogSyncException;
+import org.apache.xtable.model.catalog.CatalogTableIdentifier;
+import org.apache.xtable.model.catalog.HierarchicalTableIdentifier;
+
+@Log4j2
+public class HMSCatalogPartitionSyncOperations implements
CatalogPartitionSyncOperations {
+
+ private final IMetaStoreClient metaStoreClient;
+ private final HMSCatalogConfig catalogConfig;
+
+ public HMSCatalogPartitionSyncOperations(
+ IMetaStoreClient metaStoreClient, HMSCatalogConfig hmsCatalogConfig) {
+ this.metaStoreClient = metaStoreClient;
+ this.catalogConfig = hmsCatalogConfig;
+ }
+
+ @Override
+ public List<CatalogPartition> getAllPartitions(CatalogTableIdentifier
catalogTableIdentifier) {
+ HierarchicalTableIdentifier tableIdentifier =
+ toHierarchicalTableIdentifier(catalogTableIdentifier);
+ try {
+ return metaStoreClient
+ .listPartitions(
+ tableIdentifier.getDatabaseName(),
tableIdentifier.getTableName(), (short) -1)
+ .stream()
+ .map(p -> new CatalogPartition(p.getValues(),
p.getSd().getLocation()))
+ .collect(Collectors.toList());
+ } catch (TException e) {
+ throw new CatalogSyncException(
+ "Failed to get all partitions for table " + tableIdentifier, e);
+ }
+ }
+
+ @Override
+ public void addPartitionsToTable(
+ CatalogTableIdentifier catalogTableIdentifier, List<CatalogPartition>
partitionsToAdd) {
+ HierarchicalTableIdentifier tableIdentifier =
+ toHierarchicalTableIdentifier(catalogTableIdentifier);
+ if (partitionsToAdd.isEmpty()) {
+ log.info("No partitions to add for " + tableIdentifier);
+ return;
+ }
+ log.info("Adding partitions " + partitionsToAdd.size() + " to table " +
tableIdentifier);
+ try {
+ StorageDescriptor sd =
+ metaStoreClient
+ .getTable(tableIdentifier.getDatabaseName(),
tableIdentifier.getTableName())
+ .getSd();
+ for (List<CatalogPartition> batch :
+ CollectionUtils.batches(partitionsToAdd,
catalogConfig.getMaxPartitionsPerRequest())) {
+ List<org.apache.hadoop.hive.metastore.api.Partition> partitionList =
new ArrayList<>();
+ batch.forEach(
Review Comment:
You can also use the streams API when mapping a value in a list to another
list.
##########
xtable-hive-metastore/src/test/java/org/apache/xtable/hms/TestHMSCatalogPartitionSyncOperations.java:
##########
@@ -0,0 +1,490 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.xtable.hms;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import lombok.SneakyThrows;
+
+import org.apache.hadoop.hive.metastore.api.SerDeInfo;
+import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.thrift.TException;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import org.apache.xtable.catalog.CatalogPartition;
+import org.apache.xtable.catalog.CatalogPartitionSyncOperations;
+import org.apache.xtable.exception.CatalogSyncException;
+
+@ExtendWith(MockitoExtension.class)
+public class TestHMSCatalogPartitionSyncOperations extends
HMSCatalogSyncTestBase {
+
+ private CatalogPartitionSyncOperations mockHMSPartitionSyncOperations;
+
+ void setupCommonMocks() {
+ mockHMSPartitionSyncOperations =
+ new HMSCatalogPartitionSyncOperations(mockMetaStoreClient,
mockHMSCatalogConfig);
+ }
+
+ @SneakyThrows
+ @Test
+ void testGetAllPartitions() {
+ setupCommonMocks();
+
+ org.apache.hadoop.hive.metastore.api.Partition hivePartition1 =
Review Comment:
Let's import this class so the lines are not as long
##########
xtable-hive-metastore/src/main/java/org/apache/xtable/hms/HMSCatalogPartitionSyncOperations.java:
##########
@@ -0,0 +1,222 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.xtable.hms;
+
+import static
org.apache.xtable.catalog.CatalogUtils.toHierarchicalTableIdentifier;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import lombok.extern.log4j.Log4j2;
+
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.thrift.TException;
+
+import org.apache.hudi.common.util.CollectionUtils;
+import org.apache.hudi.exception.TableNotFoundException;
+
+import org.apache.xtable.catalog.CatalogPartition;
+import org.apache.xtable.catalog.CatalogPartitionSyncOperations;
+import org.apache.xtable.exception.CatalogSyncException;
+import org.apache.xtable.model.catalog.CatalogTableIdentifier;
+import org.apache.xtable.model.catalog.HierarchicalTableIdentifier;
+
+@Log4j2
+public class HMSCatalogPartitionSyncOperations implements
CatalogPartitionSyncOperations {
+
+ private final IMetaStoreClient metaStoreClient;
+ private final HMSCatalogConfig catalogConfig;
+
+ public HMSCatalogPartitionSyncOperations(
+ IMetaStoreClient metaStoreClient, HMSCatalogConfig hmsCatalogConfig) {
+ this.metaStoreClient = metaStoreClient;
+ this.catalogConfig = hmsCatalogConfig;
+ }
+
+ @Override
+ public List<CatalogPartition> getAllPartitions(CatalogTableIdentifier
catalogTableIdentifier) {
+ HierarchicalTableIdentifier tableIdentifier =
+ toHierarchicalTableIdentifier(catalogTableIdentifier);
+ try {
+ return metaStoreClient
+ .listPartitions(
+ tableIdentifier.getDatabaseName(),
tableIdentifier.getTableName(), (short) -1)
+ .stream()
+ .map(p -> new CatalogPartition(p.getValues(),
p.getSd().getLocation()))
+ .collect(Collectors.toList());
+ } catch (TException e) {
+ throw new CatalogSyncException(
+ "Failed to get all partitions for table " + tableIdentifier, e);
+ }
+ }
+
+ @Override
+ public void addPartitionsToTable(
+ CatalogTableIdentifier catalogTableIdentifier, List<CatalogPartition>
partitionsToAdd) {
+ HierarchicalTableIdentifier tableIdentifier =
+ toHierarchicalTableIdentifier(catalogTableIdentifier);
+ if (partitionsToAdd.isEmpty()) {
+ log.info("No partitions to add for " + tableIdentifier);
+ return;
+ }
+ log.info("Adding partitions " + partitionsToAdd.size() + " to table " +
tableIdentifier);
+ try {
+ StorageDescriptor sd =
+ metaStoreClient
+ .getTable(tableIdentifier.getDatabaseName(),
tableIdentifier.getTableName())
+ .getSd();
+ for (List<CatalogPartition> batch :
+ CollectionUtils.batches(partitionsToAdd,
catalogConfig.getMaxPartitionsPerRequest())) {
+ List<org.apache.hadoop.hive.metastore.api.Partition> partitionList =
new ArrayList<>();
Review Comment:
Can we import this class so the code is less verbose?
##########
xtable-hive-metastore/src/main/java/org/apache/xtable/hms/HMSCatalogPartitionSyncOperations.java:
##########
@@ -0,0 +1,222 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.xtable.hms;
+
+import static
org.apache.xtable.catalog.CatalogUtils.toHierarchicalTableIdentifier;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import lombok.extern.log4j.Log4j2;
+
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.thrift.TException;
+
+import org.apache.hudi.common.util.CollectionUtils;
+import org.apache.hudi.exception.TableNotFoundException;
+
+import org.apache.xtable.catalog.CatalogPartition;
+import org.apache.xtable.catalog.CatalogPartitionSyncOperations;
+import org.apache.xtable.exception.CatalogSyncException;
+import org.apache.xtable.model.catalog.CatalogTableIdentifier;
+import org.apache.xtable.model.catalog.HierarchicalTableIdentifier;
+
+@Log4j2
+public class HMSCatalogPartitionSyncOperations implements
CatalogPartitionSyncOperations {
+
+ private final IMetaStoreClient metaStoreClient;
+ private final HMSCatalogConfig catalogConfig;
+
+ public HMSCatalogPartitionSyncOperations(
+ IMetaStoreClient metaStoreClient, HMSCatalogConfig hmsCatalogConfig) {
+ this.metaStoreClient = metaStoreClient;
+ this.catalogConfig = hmsCatalogConfig;
+ }
+
+ @Override
+ public List<CatalogPartition> getAllPartitions(CatalogTableIdentifier
catalogTableIdentifier) {
+ HierarchicalTableIdentifier tableIdentifier =
+ toHierarchicalTableIdentifier(catalogTableIdentifier);
+ try {
+ return metaStoreClient
+ .listPartitions(
+ tableIdentifier.getDatabaseName(),
tableIdentifier.getTableName(), (short) -1)
+ .stream()
+ .map(p -> new CatalogPartition(p.getValues(),
p.getSd().getLocation()))
+ .collect(Collectors.toList());
+ } catch (TException e) {
+ throw new CatalogSyncException(
+ "Failed to get all partitions for table " + tableIdentifier, e);
+ }
+ }
+
+ @Override
+ public void addPartitionsToTable(
+ CatalogTableIdentifier catalogTableIdentifier, List<CatalogPartition>
partitionsToAdd) {
+ HierarchicalTableIdentifier tableIdentifier =
+ toHierarchicalTableIdentifier(catalogTableIdentifier);
+ if (partitionsToAdd.isEmpty()) {
+ log.info("No partitions to add for " + tableIdentifier);
+ return;
+ }
+ log.info("Adding partitions " + partitionsToAdd.size() + " to table " +
tableIdentifier);
+ try {
+ StorageDescriptor sd =
+ metaStoreClient
+ .getTable(tableIdentifier.getDatabaseName(),
tableIdentifier.getTableName())
+ .getSd();
+ for (List<CatalogPartition> batch :
+ CollectionUtils.batches(partitionsToAdd,
catalogConfig.getMaxPartitionsPerRequest())) {
+ List<org.apache.hadoop.hive.metastore.api.Partition> partitionList =
new ArrayList<>();
+ batch.forEach(
+ partition -> {
+ StorageDescriptor partitionSd = new StorageDescriptor();
+ partitionSd.setCols(sd.getCols());
+ partitionSd.setInputFormat(sd.getInputFormat());
+ partitionSd.setOutputFormat(sd.getOutputFormat());
+ partitionSd.setSerdeInfo(sd.getSerdeInfo());
+
+ partitionSd.setLocation(partition.getStorageLocation());
+ partitionList.add(
+ new org.apache.hadoop.hive.metastore.api.Partition(
+ partition.getValues(),
+ tableIdentifier.getDatabaseName(),
+ tableIdentifier.getTableName(),
+ 0,
+ 0,
+ partitionSd,
+ null));
+ });
+ metaStoreClient.add_partitions(partitionList, true, false);
+ log.info("Add batch partitions done: " + partitionList.size());
+ }
+ } catch (TException e) {
+ log.error("{} add partition failed", tableIdentifier, e);
+ throw new CatalogSyncException(tableIdentifier + " add partition
failed", e);
+ }
+ }
+
+ @Override
+ public void updatePartitionsToTable(
+ CatalogTableIdentifier catalogTableIdentifier, List<CatalogPartition>
changedPartitions) {
+ HierarchicalTableIdentifier tableIdentifier =
+ toHierarchicalTableIdentifier(catalogTableIdentifier);
+ try {
+ Table table =
+ metaStoreClient.getTable(
+ tableIdentifier.getDatabaseName(),
tableIdentifier.getTableName());
+ StorageDescriptor tableSd = table.getSd();
+
+ List<org.apache.hadoop.hive.metastore.api.Partition> updatedPartitions =
new ArrayList<>();
+
+ changedPartitions.forEach(
+ p -> {
+ StorageDescriptor partitionSd = new StorageDescriptor(tableSd);
+ partitionSd.setLocation(p.getStorageLocation());
+
+ org.apache.hadoop.hive.metastore.api.Partition partition =
+ new org.apache.hadoop.hive.metastore.api.Partition();
+ partition.setDbName(tableIdentifier.getDatabaseName());
+ partition.setTableName(tableIdentifier.getTableName());
+ partition.setValues(p.getValues());
+ partition.setSd(partitionSd);
+ updatedPartitions.add(partition);
+ });
+
+ // Update partitions (drop existing and add new ones with updated
locations)
Review Comment:
Can we use `alter_partitions` for this?
##########
xtable-hive-metastore/src/main/java/org/apache/xtable/hms/table/HudiHMSCatalogTableBuilder.java:
##########
@@ -0,0 +1,207 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.xtable.hms.table;
+
+import static
org.apache.xtable.catalog.CatalogUtils.toHierarchicalTableIdentifier;
+
+import java.io.IOException;
+import java.time.ZonedDateTime;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+import lombok.extern.log4j.Log4j2;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.metastore.api.FieldSchema;
+import org.apache.hadoop.hive.metastore.api.SerDeInfo;
+import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.security.UserGroupInformation;
+
+import org.apache.hudi.common.model.HoodieFileFormat;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.util.ConfigUtils;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import org.apache.xtable.catalog.CatalogTableBuilder;
+import org.apache.xtable.exception.CatalogSyncException;
+import org.apache.xtable.hms.HMSCatalogConfig;
+import org.apache.xtable.hms.HMSSchemaExtractor;
+import org.apache.xtable.hudi.HudiTableManager;
+import org.apache.xtable.hudi.catalog.HudiCatalogTableUtils;
+import org.apache.xtable.hudi.catalog.HudiInputFormatUtils;
+import org.apache.xtable.model.InternalTable;
+import org.apache.xtable.model.catalog.CatalogTableIdentifier;
+import org.apache.xtable.model.catalog.HierarchicalTableIdentifier;
+import org.apache.xtable.model.schema.InternalPartitionField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.storage.TableFormat;
+
+@Log4j2
+public class HudiHMSCatalogTableBuilder implements CatalogTableBuilder<Table,
Table> {
+
+ private final HudiTableManager hudiTableManager;
+ private final HMSSchemaExtractor schemaExtractor;
+ private final HMSCatalogConfig hmsCatalogConfig;
+
+ private HoodieTableMetaClient metaClient;
+
+ protected static final String HUDI_METADATA_CONFIG =
"hudi.metadata-listing-enabled";
+
+ public HudiHMSCatalogTableBuilder(
+ HMSCatalogConfig hmsCatalogConfig, Configuration configuration) {
+ this.hudiTableManager = HudiTableManager.of(configuration);
+ this.schemaExtractor = HMSSchemaExtractor.getInstance();
+ this.hmsCatalogConfig = hmsCatalogConfig;
+ }
+
+ @VisibleForTesting
+ HudiHMSCatalogTableBuilder(
+ HMSCatalogConfig hmsCatalogConfig,
+ HMSSchemaExtractor schemaExtractor,
+ HudiTableManager hudiTableManager,
+ HoodieTableMetaClient metaClient) {
+ this.hudiTableManager = hudiTableManager;
+ this.schemaExtractor = schemaExtractor;
+ this.metaClient = metaClient;
+ this.hmsCatalogConfig = hmsCatalogConfig;
+ }
+
+ HoodieTableMetaClient getMetaClient(String basePath) {
+ if (metaClient == null) {
+ Optional<HoodieTableMetaClient> metaClientOpt =
+ hudiTableManager.loadTableMetaClientIfExists(basePath);
+
+ if (!metaClientOpt.isPresent()) {
+ throw new CatalogSyncException(
+ "failed to get meta client since table is not present in the base
path " + basePath);
Review Comment:
Nitpick: Capitalize the first work `Failed`
##########
xtable-hive-metastore/src/main/java/org/apache/xtable/hms/table/HudiHMSCatalogTableBuilder.java:
##########
@@ -0,0 +1,207 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.xtable.hms.table;
+
+import static
org.apache.xtable.catalog.CatalogUtils.toHierarchicalTableIdentifier;
+
+import java.io.IOException;
+import java.time.ZonedDateTime;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+import lombok.extern.log4j.Log4j2;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.metastore.api.FieldSchema;
+import org.apache.hadoop.hive.metastore.api.SerDeInfo;
+import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.security.UserGroupInformation;
+
+import org.apache.hudi.common.model.HoodieFileFormat;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.util.ConfigUtils;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import org.apache.xtable.catalog.CatalogTableBuilder;
+import org.apache.xtable.exception.CatalogSyncException;
+import org.apache.xtable.hms.HMSCatalogConfig;
+import org.apache.xtable.hms.HMSSchemaExtractor;
+import org.apache.xtable.hudi.HudiTableManager;
+import org.apache.xtable.hudi.catalog.HudiCatalogTableUtils;
+import org.apache.xtable.hudi.catalog.HudiInputFormatUtils;
+import org.apache.xtable.model.InternalTable;
+import org.apache.xtable.model.catalog.CatalogTableIdentifier;
+import org.apache.xtable.model.catalog.HierarchicalTableIdentifier;
+import org.apache.xtable.model.schema.InternalPartitionField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.storage.TableFormat;
+
+@Log4j2
+public class HudiHMSCatalogTableBuilder implements CatalogTableBuilder<Table,
Table> {
+
+ private final HudiTableManager hudiTableManager;
+ private final HMSSchemaExtractor schemaExtractor;
+ private final HMSCatalogConfig hmsCatalogConfig;
+
+ private HoodieTableMetaClient metaClient;
+
+ protected static final String HUDI_METADATA_CONFIG =
"hudi.metadata-listing-enabled";
+
+ public HudiHMSCatalogTableBuilder(
+ HMSCatalogConfig hmsCatalogConfig, Configuration configuration) {
+ this.hudiTableManager = HudiTableManager.of(configuration);
+ this.schemaExtractor = HMSSchemaExtractor.getInstance();
+ this.hmsCatalogConfig = hmsCatalogConfig;
+ }
+
+ @VisibleForTesting
+ HudiHMSCatalogTableBuilder(
+ HMSCatalogConfig hmsCatalogConfig,
+ HMSSchemaExtractor schemaExtractor,
+ HudiTableManager hudiTableManager,
+ HoodieTableMetaClient metaClient) {
+ this.hudiTableManager = hudiTableManager;
+ this.schemaExtractor = schemaExtractor;
+ this.metaClient = metaClient;
+ this.hmsCatalogConfig = hmsCatalogConfig;
+ }
+
+ HoodieTableMetaClient getMetaClient(String basePath) {
+ if (metaClient == null) {
+ Optional<HoodieTableMetaClient> metaClientOpt =
+ hudiTableManager.loadTableMetaClientIfExists(basePath);
+
+ if (!metaClientOpt.isPresent()) {
+ throw new CatalogSyncException(
+ "failed to get meta client since table is not present in the base
path " + basePath);
+ }
+
+ metaClient = metaClientOpt.get();
+ }
+ return metaClient;
+ }
+
+ @Override
+ public Table getCreateTableRequest(
+ InternalTable table, CatalogTableIdentifier catalogTableIdentifier) {
+ HierarchicalTableIdentifier tableIdentifier =
+ toHierarchicalTableIdentifier(catalogTableIdentifier);
+ Table newTb = new Table();
+ newTb.setDbName(tableIdentifier.getDatabaseName());
+ newTb.setTableName(tableIdentifier.getTableName());
+ try {
+ newTb.setOwner(UserGroupInformation.getCurrentUser().getShortUserName());
+ } catch (IOException e) {
+ throw new CatalogSyncException(
+ "Failed to set owner for hms table: " +
tableIdentifier.getTableName(), e);
+ }
+
+ newTb.setCreateTime((int) ZonedDateTime.now().toEpochSecond());
+ List<String> partitionFields =
+ table.getPartitioningFields().stream()
+ .map(field -> field.getSourceField().getName())
+ .collect(Collectors.toList());
+ Map<String, String> tableProperties =
+ getTableProperties(partitionFields, table.getReadSchema());
+ newTb.setParameters(tableProperties);
+ newTb.setSd(getStorageDescriptor(table));
+ newTb.setPartitionKeys(getSchemaPartitionKeys(table));
+ return newTb;
+ }
+
+ @Override
+ public Table getUpdateTableRequest(
+ InternalTable table, Table hmsTable, CatalogTableIdentifier
tableIdentifier) {
+ Map<String, String> parameters = hmsTable.getParameters();
+ List<String> partitionFields =
+ table.getPartitioningFields().stream()
+ .map(field -> field.getSourceField().getName())
+ .collect(Collectors.toList());
+ Map<String, String> tableParameters = hmsTable.getParameters();
+ tableParameters.putAll(getTableProperties(partitionFields,
table.getReadSchema()));
+ hmsTable.setParameters(tableParameters);
+ hmsTable.setSd(getStorageDescriptor(table));
+
+ hmsTable.setParameters(parameters);
+ hmsTable.getSd().setCols(schemaExtractor.toColumns(TableFormat.HUDI,
table.getReadSchema()));
+ return hmsTable;
+ }
+
+ private Map<String, String> getTableProperties(
+ List<String> partitionFields, InternalSchema schema) {
+ Map<String, String> tableProperties = new HashMap<>();
+ tableProperties.put(HUDI_METADATA_CONFIG, "true");
+ Map<String, String> sparkTableProperties =
+ HudiCatalogTableUtils.getSparkTableProperties(
+ partitionFields, "", hmsCatalogConfig.getSchemaLengthThreshold(),
schema);
Review Comment:
The sparkVersion is passed as an empty string, should it have a value?
##########
xtable-hive-metastore/src/main/java/org/apache/xtable/hms/table/HudiHMSCatalogTableBuilder.java:
##########
@@ -0,0 +1,207 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.xtable.hms.table;
+
+import static
org.apache.xtable.catalog.CatalogUtils.toHierarchicalTableIdentifier;
+
+import java.io.IOException;
+import java.time.ZonedDateTime;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+import lombok.extern.log4j.Log4j2;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.metastore.api.FieldSchema;
+import org.apache.hadoop.hive.metastore.api.SerDeInfo;
+import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.security.UserGroupInformation;
+
+import org.apache.hudi.common.model.HoodieFileFormat;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.util.ConfigUtils;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import org.apache.xtable.catalog.CatalogTableBuilder;
+import org.apache.xtable.exception.CatalogSyncException;
+import org.apache.xtable.hms.HMSCatalogConfig;
+import org.apache.xtable.hms.HMSSchemaExtractor;
+import org.apache.xtable.hudi.HudiTableManager;
+import org.apache.xtable.hudi.catalog.HudiCatalogTableUtils;
+import org.apache.xtable.hudi.catalog.HudiInputFormatUtils;
+import org.apache.xtable.model.InternalTable;
+import org.apache.xtable.model.catalog.CatalogTableIdentifier;
+import org.apache.xtable.model.catalog.HierarchicalTableIdentifier;
+import org.apache.xtable.model.schema.InternalPartitionField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.storage.TableFormat;
+
+@Log4j2
+public class HudiHMSCatalogTableBuilder implements CatalogTableBuilder<Table,
Table> {
+
+ private final HudiTableManager hudiTableManager;
+ private final HMSSchemaExtractor schemaExtractor;
+ private final HMSCatalogConfig hmsCatalogConfig;
+
+ private HoodieTableMetaClient metaClient;
+
+ protected static final String HUDI_METADATA_CONFIG =
"hudi.metadata-listing-enabled";
+
+ public HudiHMSCatalogTableBuilder(
+ HMSCatalogConfig hmsCatalogConfig, Configuration configuration) {
+ this.hudiTableManager = HudiTableManager.of(configuration);
+ this.schemaExtractor = HMSSchemaExtractor.getInstance();
+ this.hmsCatalogConfig = hmsCatalogConfig;
+ }
+
+ @VisibleForTesting
+ HudiHMSCatalogTableBuilder(
+ HMSCatalogConfig hmsCatalogConfig,
+ HMSSchemaExtractor schemaExtractor,
+ HudiTableManager hudiTableManager,
+ HoodieTableMetaClient metaClient) {
+ this.hudiTableManager = hudiTableManager;
+ this.schemaExtractor = schemaExtractor;
+ this.metaClient = metaClient;
+ this.hmsCatalogConfig = hmsCatalogConfig;
+ }
+
+ HoodieTableMetaClient getMetaClient(String basePath) {
+ if (metaClient == null) {
+ Optional<HoodieTableMetaClient> metaClientOpt =
+ hudiTableManager.loadTableMetaClientIfExists(basePath);
+
+ if (!metaClientOpt.isPresent()) {
+ throw new CatalogSyncException(
+ "failed to get meta client since table is not present in the base
path " + basePath);
+ }
+
+ metaClient = metaClientOpt.get();
+ }
+ return metaClient;
+ }
+
+ @Override
+ public Table getCreateTableRequest(
+ InternalTable table, CatalogTableIdentifier catalogTableIdentifier) {
+ HierarchicalTableIdentifier tableIdentifier =
+ toHierarchicalTableIdentifier(catalogTableIdentifier);
+ Table newTb = new Table();
+ newTb.setDbName(tableIdentifier.getDatabaseName());
+ newTb.setTableName(tableIdentifier.getTableName());
+ try {
+ newTb.setOwner(UserGroupInformation.getCurrentUser().getShortUserName());
+ } catch (IOException e) {
+ throw new CatalogSyncException(
+ "Failed to set owner for hms table: " +
tableIdentifier.getTableName(), e);
+ }
+
+ newTb.setCreateTime((int) ZonedDateTime.now().toEpochSecond());
Review Comment:
Can you use `Instant` here instead of the ZonedDateTime
##########
xtable-hive-metastore/src/main/java/org/apache/xtable/hms/table/HudiHMSCatalogTableBuilder.java:
##########
@@ -0,0 +1,207 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.xtable.hms.table;
+
+import static
org.apache.xtable.catalog.CatalogUtils.toHierarchicalTableIdentifier;
+
+import java.io.IOException;
+import java.time.ZonedDateTime;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+import lombok.extern.log4j.Log4j2;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.metastore.api.FieldSchema;
+import org.apache.hadoop.hive.metastore.api.SerDeInfo;
+import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.security.UserGroupInformation;
+
+import org.apache.hudi.common.model.HoodieFileFormat;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.util.ConfigUtils;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import org.apache.xtable.catalog.CatalogTableBuilder;
+import org.apache.xtable.exception.CatalogSyncException;
+import org.apache.xtable.hms.HMSCatalogConfig;
+import org.apache.xtable.hms.HMSSchemaExtractor;
+import org.apache.xtable.hudi.HudiTableManager;
+import org.apache.xtable.hudi.catalog.HudiCatalogTableUtils;
+import org.apache.xtable.hudi.catalog.HudiInputFormatUtils;
+import org.apache.xtable.model.InternalTable;
+import org.apache.xtable.model.catalog.CatalogTableIdentifier;
+import org.apache.xtable.model.catalog.HierarchicalTableIdentifier;
+import org.apache.xtable.model.schema.InternalPartitionField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.storage.TableFormat;
+
+@Log4j2
+public class HudiHMSCatalogTableBuilder implements CatalogTableBuilder<Table,
Table> {
+
+ private final HudiTableManager hudiTableManager;
+ private final HMSSchemaExtractor schemaExtractor;
+ private final HMSCatalogConfig hmsCatalogConfig;
+
+ private HoodieTableMetaClient metaClient;
+
+ protected static final String HUDI_METADATA_CONFIG =
"hudi.metadata-listing-enabled";
+
+ public HudiHMSCatalogTableBuilder(
+ HMSCatalogConfig hmsCatalogConfig, Configuration configuration) {
+ this.hudiTableManager = HudiTableManager.of(configuration);
+ this.schemaExtractor = HMSSchemaExtractor.getInstance();
+ this.hmsCatalogConfig = hmsCatalogConfig;
+ }
+
+ @VisibleForTesting
+ HudiHMSCatalogTableBuilder(
+ HMSCatalogConfig hmsCatalogConfig,
+ HMSSchemaExtractor schemaExtractor,
+ HudiTableManager hudiTableManager,
+ HoodieTableMetaClient metaClient) {
+ this.hudiTableManager = hudiTableManager;
+ this.schemaExtractor = schemaExtractor;
+ this.metaClient = metaClient;
+ this.hmsCatalogConfig = hmsCatalogConfig;
+ }
+
+ HoodieTableMetaClient getMetaClient(String basePath) {
+ if (metaClient == null) {
+ Optional<HoodieTableMetaClient> metaClientOpt =
+ hudiTableManager.loadTableMetaClientIfExists(basePath);
+
+ if (!metaClientOpt.isPresent()) {
+ throw new CatalogSyncException(
+ "failed to get meta client since table is not present in the base
path " + basePath);
+ }
+
+ metaClient = metaClientOpt.get();
+ }
+ return metaClient;
+ }
+
+ @Override
+ public Table getCreateTableRequest(
+ InternalTable table, CatalogTableIdentifier catalogTableIdentifier) {
+ HierarchicalTableIdentifier tableIdentifier =
+ toHierarchicalTableIdentifier(catalogTableIdentifier);
+ Table newTb = new Table();
+ newTb.setDbName(tableIdentifier.getDatabaseName());
+ newTb.setTableName(tableIdentifier.getTableName());
+ try {
+ newTb.setOwner(UserGroupInformation.getCurrentUser().getShortUserName());
+ } catch (IOException e) {
+ throw new CatalogSyncException(
+ "Failed to set owner for hms table: " +
tableIdentifier.getTableName(), e);
+ }
+
+ newTb.setCreateTime((int) ZonedDateTime.now().toEpochSecond());
+ List<String> partitionFields =
+ table.getPartitioningFields().stream()
+ .map(field -> field.getSourceField().getName())
+ .collect(Collectors.toList());
+ Map<String, String> tableProperties =
+ getTableProperties(partitionFields, table.getReadSchema());
+ newTb.setParameters(tableProperties);
+ newTb.setSd(getStorageDescriptor(table));
+ newTb.setPartitionKeys(getSchemaPartitionKeys(table));
+ return newTb;
+ }
+
+ @Override
+ public Table getUpdateTableRequest(
+ InternalTable table, Table hmsTable, CatalogTableIdentifier
tableIdentifier) {
+ Map<String, String> parameters = hmsTable.getParameters();
+ List<String> partitionFields =
+ table.getPartitioningFields().stream()
+ .map(field -> field.getSourceField().getName())
+ .collect(Collectors.toList());
+ Map<String, String> tableParameters = hmsTable.getParameters();
+ tableParameters.putAll(getTableProperties(partitionFields,
table.getReadSchema()));
+ hmsTable.setParameters(tableParameters);
+ hmsTable.setSd(getStorageDescriptor(table));
+
+ hmsTable.setParameters(parameters);
+ hmsTable.getSd().setCols(schemaExtractor.toColumns(TableFormat.HUDI,
table.getReadSchema()));
+ return hmsTable;
+ }
+
+ private Map<String, String> getTableProperties(
+ List<String> partitionFields, InternalSchema schema) {
+ Map<String, String> tableProperties = new HashMap<>();
+ tableProperties.put(HUDI_METADATA_CONFIG, "true");
+ Map<String, String> sparkTableProperties =
+ HudiCatalogTableUtils.getSparkTableProperties(
+ partitionFields, "", hmsCatalogConfig.getSchemaLengthThreshold(),
schema);
+ tableProperties.putAll(sparkTableProperties);
+ return tableProperties;
+ }
+
+ @VisibleForTesting
+ StorageDescriptor getStorageDescriptor(InternalTable table) {
+ final StorageDescriptor storageDescriptor = new StorageDescriptor();
+ storageDescriptor.setCols(schemaExtractor.toColumns(TableFormat.HUDI,
table.getReadSchema()));
+ storageDescriptor.setLocation(table.getBasePath());
+ HoodieFileFormat fileFormat =
+
getMetaClient(table.getBasePath()).getTableConfig().getBaseFileFormat();
+ String inputFormatClassName =
HudiInputFormatUtils.getInputFormatClassName(fileFormat, false);
+ String outputFormatClassName =
HudiInputFormatUtils.getOutputFormatClassName(fileFormat);
+ String serdeClassName = HudiInputFormatUtils.getSerDeClassName(fileFormat);
+ storageDescriptor.setInputFormat(inputFormatClassName);
+ storageDescriptor.setOutputFormat(outputFormatClassName);
+ Map<String, String> serdeProperties = getSerdeProperties(false,
table.getBasePath());
+ SerDeInfo serDeInfo = new SerDeInfo();
+ serDeInfo.setSerializationLib(serdeClassName);
+ serDeInfo.setParameters(serdeProperties);
+ storageDescriptor.setSerdeInfo(serDeInfo);
+ return storageDescriptor;
+ }
+
+ private static Map<String, String> getSerdeProperties(boolean
readAsOptimized, String basePath) {
+ Map<String, String> serdeProperties = new HashMap<>();
+ serdeProperties.put(ConfigUtils.TABLE_SERDE_PATH, basePath);
+ serdeProperties.put(ConfigUtils.IS_QUERY_AS_RO_TABLE,
String.valueOf(readAsOptimized));
+ return serdeProperties;
+ }
+
+ List<FieldSchema> getSchemaPartitionKeys(InternalTable table) {
Review Comment:
Make this private?
##########
xtable-hive-metastore/src/test/java/org/apache/xtable/hms/table/TestHudiHMSCatalogTableBuilder.java:
##########
@@ -0,0 +1,143 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.xtable.hms.table;
+
+import static
org.apache.xtable.hms.table.HudiHMSCatalogTableBuilder.HUDI_METADATA_CONFIG;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.when;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.MockedStatic;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import org.apache.hudi.common.model.HoodieFileFormat;
+import org.apache.hudi.common.table.HoodieTableConfig;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+
+import org.apache.xtable.hms.HMSCatalogSyncTestBase;
+import org.apache.xtable.hms.HMSSchemaExtractor;
+import org.apache.xtable.hudi.HudiTableManager;
+import org.apache.xtable.hudi.catalog.HudiCatalogTableUtils;
+
+@ExtendWith(MockitoExtension.class)
+public class TestHudiHMSCatalogTableBuilder extends HMSCatalogSyncTestBase {
+
+ @Mock private HoodieTableMetaClient mockMetaClient;
+ @Mock private HudiTableManager mockHudiTableManager;
+ @Mock private HoodieTableConfig mockTableConfig;
+
+ private HudiHMSCatalogTableBuilder mockHudiHMSCatalogTableBuilder;
+
+ private HudiHMSCatalogTableBuilder
createMockHudiHMSCatalogSyncRequestProvider() {
+ return new HudiHMSCatalogTableBuilder(
+ mockHMSCatalogConfig,
+ HMSSchemaExtractor.getInstance(),
+ mockHudiTableManager,
+ mockMetaClient);
+ }
+
+ void setupCommonMocks() {
+ mockHudiHMSCatalogTableBuilder =
createMockHudiHMSCatalogSyncRequestProvider();
+ when(mockHMSCatalogConfig.getSchemaLengthThreshold()).thenReturn(1000);
+ }
+
+ void setupMetaClientMocks() {
+
when(mockTableConfig.getBaseFileFormat()).thenReturn(HoodieFileFormat.PARQUET);
+ when(mockMetaClient.getTableConfig()).thenReturn(mockTableConfig);
+ }
+
+ @Test
+ void testGetCreateTableInput() {
+ setupCommonMocks();
+ setupMetaClientMocks();
+
+ try (MockedStatic<HudiCatalogTableUtils> mockHudiSparkDataSourceTableUtils
=
+ mockStatic(HudiCatalogTableUtils.class)) {
Review Comment:
Let's avoid using mock static. If the logic is complicated then it should
not be in a basic util class. If it is simple then we don't need to mock it.
##########
xtable-hive-metastore/src/main/java/org/apache/xtable/hms/table/HudiHMSCatalogTableBuilder.java:
##########
@@ -0,0 +1,207 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.xtable.hms.table;
+
+import static
org.apache.xtable.catalog.CatalogUtils.toHierarchicalTableIdentifier;
+
+import java.io.IOException;
+import java.time.ZonedDateTime;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+import lombok.extern.log4j.Log4j2;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.metastore.api.FieldSchema;
+import org.apache.hadoop.hive.metastore.api.SerDeInfo;
+import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.security.UserGroupInformation;
+
+import org.apache.hudi.common.model.HoodieFileFormat;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.util.ConfigUtils;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import org.apache.xtable.catalog.CatalogTableBuilder;
+import org.apache.xtable.exception.CatalogSyncException;
+import org.apache.xtable.hms.HMSCatalogConfig;
+import org.apache.xtable.hms.HMSSchemaExtractor;
+import org.apache.xtable.hudi.HudiTableManager;
+import org.apache.xtable.hudi.catalog.HudiCatalogTableUtils;
+import org.apache.xtable.hudi.catalog.HudiInputFormatUtils;
+import org.apache.xtable.model.InternalTable;
+import org.apache.xtable.model.catalog.CatalogTableIdentifier;
+import org.apache.xtable.model.catalog.HierarchicalTableIdentifier;
+import org.apache.xtable.model.schema.InternalPartitionField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.storage.TableFormat;
+
+@Log4j2
+public class HudiHMSCatalogTableBuilder implements CatalogTableBuilder<Table,
Table> {
+
+ private final HudiTableManager hudiTableManager;
+ private final HMSSchemaExtractor schemaExtractor;
+ private final HMSCatalogConfig hmsCatalogConfig;
+
+ private HoodieTableMetaClient metaClient;
+
+ protected static final String HUDI_METADATA_CONFIG =
"hudi.metadata-listing-enabled";
+
+ public HudiHMSCatalogTableBuilder(
+ HMSCatalogConfig hmsCatalogConfig, Configuration configuration) {
+ this.hudiTableManager = HudiTableManager.of(configuration);
+ this.schemaExtractor = HMSSchemaExtractor.getInstance();
+ this.hmsCatalogConfig = hmsCatalogConfig;
+ }
+
+ @VisibleForTesting
+ HudiHMSCatalogTableBuilder(
+ HMSCatalogConfig hmsCatalogConfig,
+ HMSSchemaExtractor schemaExtractor,
+ HudiTableManager hudiTableManager,
+ HoodieTableMetaClient metaClient) {
+ this.hudiTableManager = hudiTableManager;
+ this.schemaExtractor = schemaExtractor;
+ this.metaClient = metaClient;
+ this.hmsCatalogConfig = hmsCatalogConfig;
+ }
+
+ HoodieTableMetaClient getMetaClient(String basePath) {
+ if (metaClient == null) {
+ Optional<HoodieTableMetaClient> metaClientOpt =
+ hudiTableManager.loadTableMetaClientIfExists(basePath);
+
+ if (!metaClientOpt.isPresent()) {
+ throw new CatalogSyncException(
+ "failed to get meta client since table is not present in the base
path " + basePath);
+ }
+
+ metaClient = metaClientOpt.get();
+ }
+ return metaClient;
+ }
+
+ @Override
+ public Table getCreateTableRequest(
+ InternalTable table, CatalogTableIdentifier catalogTableIdentifier) {
+ HierarchicalTableIdentifier tableIdentifier =
+ toHierarchicalTableIdentifier(catalogTableIdentifier);
+ Table newTb = new Table();
+ newTb.setDbName(tableIdentifier.getDatabaseName());
+ newTb.setTableName(tableIdentifier.getTableName());
+ try {
+ newTb.setOwner(UserGroupInformation.getCurrentUser().getShortUserName());
+ } catch (IOException e) {
+ throw new CatalogSyncException(
+ "Failed to set owner for hms table: " +
tableIdentifier.getTableName(), e);
+ }
+
+ newTb.setCreateTime((int) ZonedDateTime.now().toEpochSecond());
+ List<String> partitionFields =
+ table.getPartitioningFields().stream()
+ .map(field -> field.getSourceField().getName())
+ .collect(Collectors.toList());
+ Map<String, String> tableProperties =
+ getTableProperties(partitionFields, table.getReadSchema());
+ newTb.setParameters(tableProperties);
+ newTb.setSd(getStorageDescriptor(table));
+ newTb.setPartitionKeys(getSchemaPartitionKeys(table));
+ return newTb;
+ }
+
+ @Override
+ public Table getUpdateTableRequest(
+ InternalTable table, Table hmsTable, CatalogTableIdentifier
tableIdentifier) {
+ Map<String, String> parameters = hmsTable.getParameters();
+ List<String> partitionFields =
+ table.getPartitioningFields().stream()
+ .map(field -> field.getSourceField().getName())
+ .collect(Collectors.toList());
+ Map<String, String> tableParameters = hmsTable.getParameters();
+ tableParameters.putAll(getTableProperties(partitionFields,
table.getReadSchema()));
+ hmsTable.setParameters(tableParameters);
+ hmsTable.setSd(getStorageDescriptor(table));
+
+ hmsTable.setParameters(parameters);
+ hmsTable.getSd().setCols(schemaExtractor.toColumns(TableFormat.HUDI,
table.getReadSchema()));
+ return hmsTable;
+ }
+
+ private Map<String, String> getTableProperties(
+ List<String> partitionFields, InternalSchema schema) {
+ Map<String, String> tableProperties = new HashMap<>();
+ tableProperties.put(HUDI_METADATA_CONFIG, "true");
+ Map<String, String> sparkTableProperties =
+ HudiCatalogTableUtils.getSparkTableProperties(
+ partitionFields, "", hmsCatalogConfig.getSchemaLengthThreshold(),
schema);
+ tableProperties.putAll(sparkTableProperties);
+ return tableProperties;
+ }
+
+ @VisibleForTesting
+ StorageDescriptor getStorageDescriptor(InternalTable table) {
Review Comment:
This method is only called from within this class, should this method move
back to `private` scope and the annotation removed?
##########
xtable-hive-metastore/src/main/java/org/apache/xtable/hms/table/HudiHMSCatalogTableBuilder.java:
##########
@@ -0,0 +1,207 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.xtable.hms.table;
+
+import static
org.apache.xtable.catalog.CatalogUtils.toHierarchicalTableIdentifier;
+
+import java.io.IOException;
+import java.time.ZonedDateTime;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+import lombok.extern.log4j.Log4j2;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.metastore.api.FieldSchema;
+import org.apache.hadoop.hive.metastore.api.SerDeInfo;
+import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.security.UserGroupInformation;
+
+import org.apache.hudi.common.model.HoodieFileFormat;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.util.ConfigUtils;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import org.apache.xtable.catalog.CatalogTableBuilder;
+import org.apache.xtable.exception.CatalogSyncException;
+import org.apache.xtable.hms.HMSCatalogConfig;
+import org.apache.xtable.hms.HMSSchemaExtractor;
+import org.apache.xtable.hudi.HudiTableManager;
+import org.apache.xtable.hudi.catalog.HudiCatalogTableUtils;
+import org.apache.xtable.hudi.catalog.HudiInputFormatUtils;
+import org.apache.xtable.model.InternalTable;
+import org.apache.xtable.model.catalog.CatalogTableIdentifier;
+import org.apache.xtable.model.catalog.HierarchicalTableIdentifier;
+import org.apache.xtable.model.schema.InternalPartitionField;
+import org.apache.xtable.model.schema.InternalSchema;
+import org.apache.xtable.model.storage.TableFormat;
+
+@Log4j2
+public class HudiHMSCatalogTableBuilder implements CatalogTableBuilder<Table,
Table> {
+
+ private final HudiTableManager hudiTableManager;
+ private final HMSSchemaExtractor schemaExtractor;
+ private final HMSCatalogConfig hmsCatalogConfig;
+
+ private HoodieTableMetaClient metaClient;
+
+ protected static final String HUDI_METADATA_CONFIG =
"hudi.metadata-listing-enabled";
+
+ public HudiHMSCatalogTableBuilder(
+ HMSCatalogConfig hmsCatalogConfig, Configuration configuration) {
+ this.hudiTableManager = HudiTableManager.of(configuration);
+ this.schemaExtractor = HMSSchemaExtractor.getInstance();
+ this.hmsCatalogConfig = hmsCatalogConfig;
+ }
+
+ @VisibleForTesting
+ HudiHMSCatalogTableBuilder(
+ HMSCatalogConfig hmsCatalogConfig,
+ HMSSchemaExtractor schemaExtractor,
+ HudiTableManager hudiTableManager,
+ HoodieTableMetaClient metaClient) {
+ this.hudiTableManager = hudiTableManager;
+ this.schemaExtractor = schemaExtractor;
+ this.metaClient = metaClient;
+ this.hmsCatalogConfig = hmsCatalogConfig;
+ }
+
+ HoodieTableMetaClient getMetaClient(String basePath) {
+ if (metaClient == null) {
+ Optional<HoodieTableMetaClient> metaClientOpt =
+ hudiTableManager.loadTableMetaClientIfExists(basePath);
+
+ if (!metaClientOpt.isPresent()) {
+ throw new CatalogSyncException(
+ "failed to get meta client since table is not present in the base
path " + basePath);
+ }
+
+ metaClient = metaClientOpt.get();
+ }
+ return metaClient;
+ }
+
+ @Override
+ public Table getCreateTableRequest(
+ InternalTable table, CatalogTableIdentifier catalogTableIdentifier) {
+ HierarchicalTableIdentifier tableIdentifier =
+ toHierarchicalTableIdentifier(catalogTableIdentifier);
+ Table newTb = new Table();
+ newTb.setDbName(tableIdentifier.getDatabaseName());
+ newTb.setTableName(tableIdentifier.getTableName());
+ try {
+ newTb.setOwner(UserGroupInformation.getCurrentUser().getShortUserName());
+ } catch (IOException e) {
+ throw new CatalogSyncException(
+ "Failed to set owner for hms table: " +
tableIdentifier.getTableName(), e);
+ }
+
+ newTb.setCreateTime((int) ZonedDateTime.now().toEpochSecond());
+ List<String> partitionFields =
+ table.getPartitioningFields().stream()
+ .map(field -> field.getSourceField().getName())
+ .collect(Collectors.toList());
+ Map<String, String> tableProperties =
+ getTableProperties(partitionFields, table.getReadSchema());
+ newTb.setParameters(tableProperties);
+ newTb.setSd(getStorageDescriptor(table));
+ newTb.setPartitionKeys(getSchemaPartitionKeys(table));
+ return newTb;
+ }
+
+ @Override
+ public Table getUpdateTableRequest(
+ InternalTable table, Table hmsTable, CatalogTableIdentifier
tableIdentifier) {
+ Map<String, String> parameters = hmsTable.getParameters();
+ List<String> partitionFields =
+ table.getPartitioningFields().stream()
+ .map(field -> field.getSourceField().getName())
+ .collect(Collectors.toList());
+ Map<String, String> tableParameters = hmsTable.getParameters();
+ tableParameters.putAll(getTableProperties(partitionFields,
table.getReadSchema()));
+ hmsTable.setParameters(tableParameters);
+ hmsTable.setSd(getStorageDescriptor(table));
+
+ hmsTable.setParameters(parameters);
+ hmsTable.getSd().setCols(schemaExtractor.toColumns(TableFormat.HUDI,
table.getReadSchema()));
+ return hmsTable;
+ }
+
+ private Map<String, String> getTableProperties(
Review Comment:
Can we have this take in the InternalTable object so we can move the
computation of the partition fields into this shared code as well? Both callers
to this method run this right before:
```
List<String> partitionFields =
table.getPartitioningFields().stream()
.map(field -> field.getSourceField().getName())
.collect(Collectors.toList());
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@xtable.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
