-
Notifications
You must be signed in to change notification settings - Fork 15
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
* Initial commit with flakey test * Updates to test * Updates to test * Adding column in between two columns * Some changes to use Parquet * Test now failing successfully * Creating AlterTableService and operations * Fixing integration test * Fixing integration test * Fixing unit tests and copyright * Some fixes * Adding DropTableService which removes custom table parameters before dropping * Adding changelog and other bits * Added catch for non existent table * Fixing some bits, renaming, adding external key to table when dropping * Undoing unnecessary changes * Fixes * Fixing pom * Fixing pom and test names * Minor change * Fixing tests
- Loading branch information
Max Jacobs
authored
Mar 16, 2020
1 parent
61c91b3
commit f6f5389
Showing
17 changed files
with
1,108 additions
and
25 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
83 changes: 83 additions & 0 deletions
83
...in-core/src/main/java/com/hotels/bdp/circustrain/core/replica/hive/AlterTableService.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,83 @@ | ||
| /** | ||
| * Copyright (C) 2016-2020 Expedia, Inc. | ||
| * | ||
| * Licensed 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 com.hotels.bdp.circustrain.core.replica.hive; | ||
|
|
||
| import java.util.List; | ||
| import java.util.Map; | ||
| import java.util.function.Function; | ||
| import java.util.stream.Collectors; | ||
|
|
||
| import org.apache.hadoop.hive.metastore.api.FieldSchema; | ||
| import org.apache.hadoop.hive.metastore.api.Table; | ||
| import org.apache.thrift.TException; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| import com.hotels.hcommon.hive.metastore.client.api.CloseableMetaStoreClient; | ||
|
|
||
| public class AlterTableService { | ||
|
|
||
| private static final Logger LOG = LoggerFactory.getLogger(AlterTableService.class); | ||
|
|
||
| private DropTableService dropTableService; | ||
| private CopyPartitionsOperation copyPartitionsOperation; | ||
| private RenameTableOperation renameTableOperation; | ||
|
|
||
| public AlterTableService( | ||
| DropTableService dropTableService, | ||
| CopyPartitionsOperation copyPartitionsOperation, | ||
| RenameTableOperation renameTableOperation) { | ||
| this.dropTableService = dropTableService; | ||
| this.copyPartitionsOperation = copyPartitionsOperation; | ||
| this.renameTableOperation = renameTableOperation; | ||
| } | ||
|
|
||
| public void alterTable(CloseableMetaStoreClient client, Table oldTable, Table newTable) throws TException { | ||
| List<FieldSchema> oldColumns = oldTable.getSd().getCols(); | ||
| List<FieldSchema> newColumns = newTable.getSd().getCols(); | ||
| if (hasAnyChangedColumns(oldColumns, newColumns)) { | ||
| LOG | ||
| .info("Found columns that have changed type, attempting to recreate target table with the new columns." | ||
| + "Old columns: {}, new columns: {}", oldColumns, newColumns); | ||
| Table tempTable = new Table(newTable); | ||
| String tempName = newTable.getTableName() + "_temp"; | ||
| tempTable.setTableName(tempName); | ||
| try { | ||
| client.createTable(tempTable); | ||
| copyPartitionsOperation.execute(client, newTable, tempTable); | ||
| renameTableOperation.execute(client, tempTable, newTable); | ||
| } finally { | ||
| dropTableService.removeTableParamsAndDrop(client, tempTable.getDbName(), tempName); | ||
| } | ||
| } else { | ||
| client.alter_table(newTable.getDbName(), newTable.getTableName(), newTable); | ||
| } | ||
| } | ||
|
|
||
| private boolean hasAnyChangedColumns(List<FieldSchema> oldColumns, List<FieldSchema> newColumns) { | ||
| Map<String, FieldSchema> oldColumnsMap = oldColumns.stream() | ||
| .collect(Collectors.toMap(FieldSchema::getName, Function.identity())); | ||
| for (FieldSchema newColumn : newColumns) { | ||
| if (oldColumnsMap.containsKey(newColumn.getName())) { | ||
| FieldSchema oldColumn = oldColumnsMap.get(newColumn.getName()); | ||
| if (!oldColumn.getType().equals(newColumn.getType())) { | ||
| return true; | ||
| } | ||
| } | ||
| } | ||
| return false; | ||
| } | ||
| } |
76 changes: 76 additions & 0 deletions
76
...e/src/main/java/com/hotels/bdp/circustrain/core/replica/hive/CopyPartitionsOperation.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,76 @@ | ||
| /** | ||
| * Copyright (C) 2016-2020 Expedia, Inc. | ||
| * | ||
| * Licensed 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 com.hotels.bdp.circustrain.core.replica.hive; | ||
|
|
||
| import java.util.ArrayList; | ||
| import java.util.List; | ||
|
|
||
| import org.apache.hadoop.hive.metastore.api.Partition; | ||
| import org.apache.hadoop.hive.metastore.api.StorageDescriptor; | ||
| import org.apache.hadoop.hive.metastore.api.Table; | ||
| import org.apache.thrift.TException; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| import com.google.common.annotations.VisibleForTesting; | ||
|
|
||
| import com.hotels.hcommon.hive.metastore.client.api.CloseableMetaStoreClient; | ||
| import com.hotels.hcommon.hive.metastore.iterator.PartitionIterator; | ||
|
|
||
| public class CopyPartitionsOperation { | ||
|
|
||
| private static final Logger LOG = LoggerFactory.getLogger(CopyPartitionsOperation.class); | ||
| private static final short DEFAULT_BATCH_SIZE = 1000; | ||
|
|
||
| private short partitionBatchSize; | ||
|
|
||
| public CopyPartitionsOperation() { | ||
| this(DEFAULT_BATCH_SIZE); | ||
| } | ||
|
|
||
| @VisibleForTesting | ||
| CopyPartitionsOperation(short partitionBatchSize) { | ||
| this.partitionBatchSize = partitionBatchSize; | ||
| } | ||
|
|
||
| /** | ||
| * Copies partitions from oldTable to newTable, partitions copied are modified to take the schema of newTable | ||
| */ | ||
| public void execute(CloseableMetaStoreClient client, Table oldTable, Table newTable) throws TException { | ||
| int count = 0; | ||
| String databaseName = newTable.getDbName(); | ||
| String tableName = newTable.getTableName(); | ||
| PartitionIterator partitionIterator = new PartitionIterator(client, oldTable, partitionBatchSize); | ||
| while (partitionIterator.hasNext()) { | ||
| List<Partition> batch = new ArrayList<>(); | ||
| for (int i = 0; i < partitionBatchSize && partitionIterator.hasNext(); i++) { | ||
| Partition partition = partitionIterator.next(); | ||
| count++; | ||
| Partition copy = new Partition(partition); | ||
| copy.setDbName(databaseName); | ||
| copy.setTableName(tableName); | ||
| StorageDescriptor sd = new StorageDescriptor(partition.getSd()); | ||
| sd.setCols(newTable.getSd().getCols()); | ||
| copy.setSd(sd); | ||
| batch.add(copy); | ||
| } | ||
| LOG.info("Copying batch of size {} to {}.{}", batch.size(), databaseName, tableName); | ||
| client.add_partitions(batch); | ||
| } | ||
| LOG.info("Copied {} partitions to {}.{}", count, databaseName, tableName); | ||
| } | ||
|
|
||
| } |
67 changes: 67 additions & 0 deletions
67
...ain-core/src/main/java/com/hotels/bdp/circustrain/core/replica/hive/DropTableService.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,67 @@ | ||
| /** | ||
| * Copyright (C) 2016-2020 Expedia, Inc. | ||
| * | ||
| * Licensed 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 com.hotels.bdp.circustrain.core.replica.hive; | ||
|
|
||
| import java.util.Collections; | ||
| import java.util.Map; | ||
|
|
||
| import org.apache.commons.collections.map.CaseInsensitiveMap; | ||
| import org.apache.hadoop.hive.metastore.api.NoSuchObjectException; | ||
| import org.apache.hadoop.hive.metastore.api.Table; | ||
| import org.apache.thrift.TException; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| import com.hotels.hcommon.hive.metastore.client.api.CloseableMetaStoreClient; | ||
|
|
||
| public class DropTableService { | ||
|
|
||
| private static final Logger LOG = LoggerFactory.getLogger(DropTableService.class); | ||
| private static final String EXTERNAL_KEY = "EXTERNAL"; | ||
| private static final String IS_EXTERNAL = "TRUE"; | ||
|
|
||
| /** | ||
| * Removes all parameters from a table before dropping the table. | ||
| */ | ||
| public void removeTableParamsAndDrop( | ||
| CloseableMetaStoreClient client, | ||
| String databaseName, | ||
| String tableName) throws TException { | ||
| Table table; | ||
| try { | ||
| table = client.getTable(databaseName, tableName); | ||
| } catch (NoSuchObjectException e) { | ||
| return; | ||
| } | ||
| Map<String, String> tableParameters = table.getParameters(); | ||
| if (tableParameters != null && !tableParameters.isEmpty()) { | ||
| if (isExternal(tableParameters)) { | ||
| table.setParameters(Collections.singletonMap(EXTERNAL_KEY, IS_EXTERNAL)); | ||
| } else { | ||
| table.setParameters(Collections.emptyMap()); | ||
| } | ||
| client.alter_table(databaseName, tableName, table); | ||
| } | ||
| LOG | ||
| .info("Dropping table '{}.{}'.", table.getDbName(), table.getTableName()); | ||
| client.dropTable(table.getDbName(), table.getTableName(), false, true); | ||
| } | ||
|
|
||
| private boolean isExternal(Map<String, String> tableParameters) { | ||
| CaseInsensitiveMap caseInsensitiveParams = new CaseInsensitiveMap(tableParameters); | ||
| return IS_EXTERNAL.equalsIgnoreCase((String) caseInsensitiveParams.get(EXTERNAL_KEY)); | ||
| } | ||
| } |
Oops, something went wrong.