import org.apache.hadoop.hive.metastore.api.Partition; //导入依赖的package包/类
PartitionedTablePathResolver(IMetaStoreClient metastore, Table table)
    throws NoSuchObjectException, MetaException, TException {
  this.metastore = metastore;
  this.table = table;
  LOG.debug("Table '{}' is partitioned", Warehouse.getQualifiedName(table));
  tableBaseLocation = locationAsPath(table);
  List<Partition> onePartition = metastore.listPartitions(table.getDbName(), table.getTableName(), (short) 1);
  if (onePartition.isEmpty()) {
    LOG.warn("Table '{}' has no partitions, perhaps you can simply delete: {}.", Warehouse.getQualifiedName(table),
        tableBaseLocation);
    throw new ConfigurationException();
  }
  Path partitionLocation = locationAsPath(onePartition.get(0));
  int branches = partitionLocation.depth() - tableBaseLocation.depth();
  String globSuffix = StringUtils.repeat("*", "/", branches);
  globPath = new Path(tableBaseLocation, globSuffix);
}
 

import org.apache.hadoop.hive.metastore.api.Partition; //导入依赖的package包/类
@Test
public void addPartition() throws Exception {
  String tableName = "my_table";
  createPartitionedTable(DATABASE, tableName);

  try (Connection connection = DriverManager.getConnection(server.connectionURL());
      Statement statement = connection.createStatement()) {
    String addPartitionHql = String.format("ALTER TABLE %s.%s ADD PARTITION (partcol=1)", DATABASE, tableName);
    statement.execute(addPartitionHql);
  }

  HiveMetaStoreClient client = server.newClient();
  try {
    List<Partition> partitions = client.listPartitions(DATABASE, tableName, (short) -1);
    assertThat(partitions.size(), is(1));
    assertThat(partitions.get(0).getDbName(), is(DATABASE));
    assertThat(partitions.get(0).getTableName(), is(tableName));
    assertThat(partitions.get(0).getValues(), is(Arrays.asList("1")));
    assertThat(partitions.get(0).getSd().getLocation(),
        is(String.format("file:%s/%s/%s/partcol=1", server.temporaryFolder.getRoot(), DATABASE, tableName)));
  } finally {
    client.close();
  }
}
 

import org.apache.hadoop.hive.metastore.api.Partition; //导入依赖的package包/类
@Override
public Set<Path> getMetastorePaths(short batchSize, int expectedPathCount)
  throws NoSuchObjectException, MetaException, TException {
  Set<Path> metastorePaths = new HashSet<>(expectedPathCount);
  PartitionIterator partitionIterator = new PartitionIterator(metastore, table, batchSize);
  while (partitionIterator.hasNext()) {
    Partition partition = partitionIterator.next();
    Path location = PathUtils.normalise(locationAsPath(partition));
    if (!location.toString().toLowerCase().startsWith(tableBaseLocation.toString().toLowerCase())) {
      LOG.error("Check your configuration: '{}' does not appear to be part of '{}'.", location, tableBaseLocation);
      throw new ConfigurationException();
    }
    metastorePaths.add(location);
  }
  return metastorePaths;
}
 

import org.apache.hadoop.hive.metastore.api.Partition; //导入依赖的package包/类
@JsonCreator
public HivePartition(@JsonProperty("values") List<String> values, @JsonProperty("tableName") String tableName, @JsonProperty("dbName") String dbName, @JsonProperty("createTime") int createTime,
                     @JsonProperty("lastAccessTime") int lastAccessTime,  @JsonProperty("sd") StorageDescriptorWrapper sd,
                     @JsonProperty("parameters") Map<String, String> parameters
) {
  this.values = values;
  this.tableName = tableName;
  this.dbName = dbName;
  this.createTime = createTime;
  this.lastAccessTime = lastAccessTime;
  this.sd = sd;
  this.parameters = parameters;

  StorageDescriptor sdUnwrapped = sd.getSd();
  this.partition = new org.apache.hadoop.hive.metastore.api.Partition(values, tableName, dbName, createTime, lastAccessTime, sdUnwrapped, parameters);
}
 

import org.apache.hadoop.hive.metastore.api.Partition; //导入依赖的package包/类
@Override
public boolean hasNext() {
  if (batch.hasNext()) {
    return true;
  }
  if (partitionNames.hasNext()) {
    List<String> names = partitionNames.next();
    try {
      List<Partition> partitions = metastore.getPartitionsByNames(table.getDbName(), table.getTableName(), names);
      count += partitions.size();
      LOG.debug("Retrieved {} partitions, total: {}.", partitions.size(), count);
      batch = partitions.iterator();
    } catch (TException e) {
      throw new RuntimeException(e);
    }
  }
  return batch.hasNext();
}
 

import org.apache.hadoop.hive.metastore.api.Partition; //导入依赖的package包/类
@Test
public void tablesAreDifferent() throws Exception {
  Table sourceTable = catalog.client().getTable(DATABASE, SOURCE_TABLE);
  sourceTable.getParameters().put("com.company.team", "value");
  catalog.client().alter_table(DATABASE, SOURCE_TABLE, sourceTable);

  // Reload table object
  sourceTable = catalog.client().getTable(DATABASE, SOURCE_TABLE);
  Table replicaTable = catalog.client().getTable(DATABASE, REPLICA_TABLE);

  HiveDifferences
      .builder(diffListener)
      .comparatorRegistry(comparatorRegistry)
      .source(configuration, sourceTable, new PartitionIterator(catalog.client(), sourceTable, PARTITION_BATCH_SIZE))
      .replica(Optional.of(replicaTable),
          Optional.of(new BufferedPartitionFetcher(catalog.client(), replicaTable, PARTITION_BATCH_SIZE)))
      .checksumFunction(checksumFunction)
      .build()
      .run();
  verify(diffListener, times(1)).onChangedTable(anyList());
  verify(diffListener, never()).onNewPartition(anyString(), any(Partition.class));
  verify(diffListener, never()).onChangedPartition(anyString(), any(Partition.class), anyList());
  verify(diffListener, never()).onDataChanged(anyString(), any(Partition.class));
}
 

import org.apache.hadoop.hive.metastore.api.Partition; //导入依赖的package包/类
HdfsSnapshotLocationManager(
    HiveConf sourceHiveConf,
    String eventId,
    Table sourceTable,
    boolean snapshotsDisabled,
    String tableBasePath,
    SourceCatalogListener sourceCatalogListener) throws IOException {
  this(sourceHiveConf, eventId, sourceTable, Collections.<Partition> emptyList(), snapshotsDisabled, tableBasePath,
      sourceCatalogListener);
}
 

import org.apache.hadoop.hive.metastore.api.Partition; //导入依赖的package包/类
public SourceLocationManager getLocationManager(
    Table table,
    List<Partition> partitions,
    String eventId,
    Map<String, Object> copierOptions)
  throws IOException {
  if (MetaStoreUtils.isView(table)) {
    return new ViewLocationManager();
  }
  HdfsSnapshotLocationManager hdfsSnapshotLocationManager = new HdfsSnapshotLocationManager(getHiveConf(), eventId,
      table, partitions, snapshotsDisabled, sourceTableLocation, sourceCatalogListener);
  boolean ignoreMissingFolder = MapUtils.getBooleanValue(copierOptions,
      CopierOptions.IGNORE_MISSING_PARTITION_FOLDER_ERRORS, false);
  if (ignoreMissingFolder) {
    return new FilterMissingPartitionsLocationManager(hdfsSnapshotLocationManager, getHiveConf());
  }
  return hdfsSnapshotLocationManager;
}
 

import org.apache.hadoop.hive.metastore.api.Partition; //导入依赖的package包/类
private void setupHiveTables() throws TException, IOException {
  List<FieldSchema> partitionKeys = Lists.newArrayList(newFieldSchema("p1"), newFieldSchema("p2"));

  File tableLocation = new File("db1", "table1");
  StorageDescriptor sd = newStorageDescriptor(tableLocation, "col0");
  table1 = newTable("table1", "db1", partitionKeys, sd);
  Partition partition1 = newPartition(table1, "value1", "value2");
  Partition partition2 = newPartition(table1, "value11", "value22");
  table1Partitions = Arrays.asList(partition1, partition2); //
  table1PartitionNames = Arrays.asList(Warehouse.makePartName(partitionKeys, partition1.getValues()),
      Warehouse.makePartName(partitionKeys, partition2.getValues()));

  File tableLocation2 = new File("db2", "table2");
  StorageDescriptor sd2 = newStorageDescriptor(tableLocation2, "col0");
  table2 = newTable("table2", "db2", partitionKeys, sd2);
}
 

import org.apache.hadoop.hive.metastore.api.Partition; //导入依赖的package包/类
@Test
public void noMatchingPartitions() throws Exception {
  PartitionsAndStatistics emptyPartitionsAndStats = new PartitionsAndStatistics(sourceTable.getPartitionKeys(),
      Collections.<Partition> emptyList(), Collections.<String, List<ColumnStatisticsObj>> emptyMap());
  when(source.getPartitions(sourceTable, PARTITION_PREDICATE, MAX_PARTITIONS)).thenReturn(emptyPartitionsAndStats);
  when(source.getLocationManager(sourceTable, Collections.<Partition> emptyList(), EVENT_ID, copierOptions))
      .thenReturn(sourceLocationManager);

  PartitionedTableReplication replication = new PartitionedTableReplication(DATABASE, TABLE, partitionPredicate,
      source, replica, copierFactoryManager, eventIdFactory, targetTableLocation, DATABASE, TABLE, copierOptions,
      listener);
  replication.replicate();

  verifyZeroInteractions(copier);
  InOrder replicationOrder = inOrder(sourceLocationManager, replica, replicaLocationManager, listener);
  replicationOrder.verify(replica).validateReplicaTable(DATABASE, TABLE);
  replicationOrder.verify(replica).updateMetadata(EVENT_ID, sourceTableAndStatistics, DATABASE, TABLE,
      replicaLocationManager);
}
 

import org.apache.hadoop.hive.metastore.api.Partition; //导入依赖的package包/类
@Test
public void noMatchingPartitions() throws Exception {
  PartitionsAndStatistics emptyPartitionsAndStats = new PartitionsAndStatistics(sourceTable.getPartitionKeys(),
      Collections.<Partition> emptyList(), Collections.<String, List<ColumnStatisticsObj>> emptyMap());
  when(source.getPartitions(sourceTable, PARTITION_PREDICATE, MAX_PARTITIONS)).thenReturn(emptyPartitionsAndStats);
  when(source.getLocationManager(sourceTable, Collections.<Partition> emptyList(), EVENT_ID, copierOptions))
      .thenReturn(sourceLocationManager);

  PartitionedTableMetadataUpdateReplication replication = new PartitionedTableMetadataUpdateReplication(DATABASE,
      TABLE, partitionPredicate, source, replica, eventIdFactory, replicaLocation, DATABASE, TABLE);
  replication.replicate();

  verify(replica).validateReplicaTable(DATABASE, TABLE);
  verify(replica).updateMetadata(eq(EVENT_ID), eq(sourceTableAndStatistics), eq(DATABASE), eq(TABLE),
      any(MetadataUpdateReplicaLocationManager.class));
}
 

import org.apache.hadoop.hive.metastore.api.Partition; //导入依赖的package包/类
@Test
public void typical() throws Exception {
  List<FieldSchema> partitionKeys = Lists.newArrayList(newFieldSchema("a"), newFieldSchema("c"));
  Table table = newTable("t1", "db1", partitionKeys, newStorageDescriptor(new File("bla"), "col1"));
  List<Partition> partitions = Lists.newArrayList(newPartition(table, "b", "d"));
  statisticsPerPartitionName.put("a=b/c=d", columnStats);

  PartitionsAndStatistics partitionsAndStatistics = new PartitionsAndStatistics(partitionKeys, partitions,
      statisticsPerPartitionName);
  List<String> expectedName = Lists.newArrayList("a=b/c=d");

  assertThat(partitionsAndStatistics.getPartitionNames(), is(expectedName));
  assertThat(partitionsAndStatistics.getPartitions(), is(partitions));
  ColumnStatisticsDesc statsDesc = new ColumnStatisticsDesc(false, "db1", "t1");
  statsDesc.setPartName("a=b/c=d");
  ColumnStatistics expectedStats = new ColumnStatistics(statsDesc, columnStats);
  assertThat(partitionsAndStatistics.getStatisticsForPartition(partitions.get(0)), is(expectedStats));
}
 

import org.apache.hadoop.hive.metastore.api.Partition; //导入依赖的package包/类
private static List<PartitionValue> getPartitions(Table table, Partition partition) {
  if(partition == null){
    return Collections.emptyList();
  }

  final List<String> partitionValues = partition.getValues();
  final List<PartitionValue> output = Lists.newArrayList();
  final List<FieldSchema> partitionKeys = table.getPartitionKeys();
  for(int i =0; i < partitionKeys.size(); i++){
    PartitionValue value = getPartitionValue(partitionKeys.get(i), partitionValues.get(i));
    if(value != null){
      output.add(value);
    }
  }
  return output;
}
 

import org.apache.hadoop.hive.metastore.api.Partition; //导入依赖的package包/类
@Test
public void replicaTableDoesNotExist() {
  hiveDifferences = HiveDifferences
      .builder(diffListener)
      .comparatorRegistry(comparatorRegistry)
      .source(sourceConfiguration, sourceTable, sourcePartitionIterable)
      .replica(Optional.<Table> absent(), Optional.<PartitionFetcher> absent())
      .checksumFunction(checksumFunction)
      .build();
  hiveDifferences.run();

  InOrder inOrder = inOrder(diffListener);
  inOrder.verify(diffListener).onDiffStart(any(TableAndMetadata.class), any(Optional.class));
  verify(diffListener, never()).onChangedTable(anyList());
  inOrder.verify(diffListener, times(1)).onNewPartition(anyString(), any(Partition.class));
  verify(diffListener, never()).onChangedPartition(anyString(), any(Partition.class), anyList());
  verify(diffListener, never()).onDataChanged(anyString(), any(Partition.class));
  inOrder.verify(diffListener).onDiffEnd();
}
 

import org.apache.hadoop.hive.metastore.api.Partition; //导入依赖的package包/类
@Test
public void dropPartition() throws Exception {
  String tableName = "my_table";
  HiveMetaStoreClient client = server.newClient();

  try {
    Table table = createPartitionedTable(DATABASE, tableName);

    Partition partition = new Partition();
    partition.setDbName(DATABASE);
    partition.setTableName(tableName);
    partition.setValues(Arrays.asList("1"));
    partition.setSd(new StorageDescriptor(table.getSd()));
    partition.getSd().setLocation(
        String.format("file:%s/%s/%s/partcol=1", server.temporaryFolder.getRoot(), DATABASE, tableName));
    client.add_partition(partition);

    try (Connection connection = DriverManager.getConnection(server.connectionURL());
        Statement statement = connection.createStatement()) {
      String addPartitionHql = String.format("ALTER TABLE %s.%s DROP PARTITION (partcol=1)", DATABASE, tableName);
      statement.execute(addPartitionHql);
    }

    List<Partition> partitions = client.listPartitions(DATABASE, tableName, (short) -1);
    assertThat(partitions.size(), is(0));
  } finally {
    client.close();
  }
}
 

import org.apache.hadoop.hive.metastore.api.Partition; //导入依赖的package包/类
@Test
public void onChangedPartition() throws Exception {
  Partition partition1 = new Partition(Lists.newArrayList("val1", "val2"), DB, TABLE, 1, 1, null, null);
  Partition partition2 = new Partition(Lists.newArrayList("val11", "val22"), DB, TABLE, 1, 1, null, null);
  listener.onDiffStart(source, replica);
  listener.onChangedPartition("p1", partition1, differences);
  listener.onChangedPartition("p2", partition2, differences);
  assertThat(listener.getPartitionSpecFilter(), is("(p1='val1' AND p2=val2) OR (p1='val11' AND p2=val22)"));
}
 

import org.apache.hadoop.hive.metastore.api.Partition; //导入依赖的package包/类
@Test
public void replicaPartitionHasChanged() throws Exception {
  Partition replicaPartition1 = catalog.client().getPartition(DATABASE, REPLICA_TABLE, "part=1");
  replicaPartition1.getSd().getCols().add(BAZ_COL);
  catalog.client().alter_partition(DATABASE, REPLICA_TABLE, replicaPartition1);

  Table sourceTable = catalog.client().getTable(DATABASE, SOURCE_TABLE);
  Table replicaTable = catalog.client().getTable(DATABASE, REPLICA_TABLE);

  HiveDifferences
      .builder(diffListener)
      .comparatorRegistry(comparatorRegistry)
      .source(configuration, sourceTable, new PartitionIterator(catalog.client(), sourceTable, PARTITION_BATCH_SIZE))
      .replica(Optional.of(replicaTable),
          Optional.of(new BufferedPartitionFetcher(catalog.client(), replicaTable, PARTITION_BATCH_SIZE)))
      .checksumFunction(checksumFunction)
      .build()
      .run();
  verify(diffListener, never()).onChangedTable(anyList());
  verify(diffListener, never()).onNewPartition(anyString(), any(Partition.class));
  verify(diffListener, times(1)).onChangedPartition("part=1",
      catalog.client().getPartition(DATABASE, SOURCE_TABLE, "part=1"),
      Arrays.<Diff<Object, Object>> asList(new BaseDiff<Object, Object>(
          "Collection partition.sd.cols of class java.util.ArrayList has different size: left.size()=2 and right.size()=3",
          Arrays.asList(FOO_COL, BAR_COL), Arrays.asList(FOO_COL, BAR_COL, BAZ_COL))));
  verify(diffListener, never()).onDataChanged(anyString(), any(Partition.class));
}
 

import org.apache.hadoop.hive.metastore.api.Partition; //导入依赖的package包/类
@Override
public void run() throws CircusTrainException {
  out.println(String.format("Source catalog:        %s", source.getName()));
  out.println(String.format("Source MetaStore URIs: %s", source.getMetaStoreUris()));
  out.println(String.format("Source table:          %s", Warehouse.getQualifiedName(sourceTable)));
  out.println(String.format("Partition expression:  %s", partitionFilter));

  String parsedPartitionFilter = partitionPredicate.getPartitionPredicate();
  if (!Objects.equals(partitionFilter, parsedPartitionFilter)) {
    LOG.info("Evaluated expression to: {}", parsedPartitionFilter);
  }
  try {
    LOG.info("Executing filter with limit {} on: {}:{} ({})", partitionLimit, source.getName(),
        Warehouse.getQualifiedName(sourceTable), source.getMetaStoreUris());
    PartitionsAndStatistics partitions = source.getPartitions(sourceTable, parsedPartitionFilter, partitionLimit);
    LOG.info("Retrieved {} partition(s):", partitions.getPartitions().size());
    SortedSet<Partition> sorted = new TreeSet<>(PARTITION_COMPARATOR);
    sorted.addAll(partitions.getPartitions());
    List<List<String>> vals = new ArrayList<>();
    for (Partition partition : sorted) {
      vals.add(partition.getValues());
      LOG.info("{}", partition.getValues());
    }
    out.println(String.format("Partition filter:      %s", parsedPartitionFilter));
    out.println(String.format("Partition limit:       %s", partitionLimit));
    out.println(String.format("Partition(s) fetched:  %s", vals));
  } catch (TException e) {
    throw new CircusTrainException("Could not fetch partitions for filter: '" + parsedPartitionFilter + "'.", e);
  }
}
 

import org.apache.hadoop.hive.metastore.api.Partition; //导入依赖的package包/类
@Override
public Partition fetch(String partitionName) {
  int partitionPosition = partitionNames.indexOf(partitionName);
  if (partitionPosition < 0) {
    throw new PartitionNotFoundException("Unknown partition " + partitionName);
  }

  if (!buffer.containsKey(partitionName)) {
    bufferPartitions(partitionPosition);
  }

  return buffer.get(partitionName);
}
 

import org.apache.hadoop.hive.metastore.api.Partition; //导入依赖的package包/类
public static Partition newPartition() {
  Partition partition = new Partition();
  StorageDescriptor sd = new StorageDescriptor();
  SerDeInfo info = new SerDeInfo();
  info.setParameters(new HashMap<String, String>());
  sd.setSerdeInfo(info);
  partition.setSd(sd);
  partition.setParameters(new HashMap<String, String>());
  return partition;
}