import org.pentaho.di.partition.PartitionSchema; //导入依赖的package包/类
public TransDialog(Shell parent, int style, TransMeta transMeta, Repository rep)
{
    super(parent, style);
    this.log      = LogWriter.getInstance();
    this.props    = PropsUI.getInstance();
    this.transMeta    = transMeta;
    this.rep      = rep;
    
    this.newDirectory = null;
    
    schemas = new ArrayList<PartitionSchema>();
    for (int i=0;i<transMeta.getPartitionSchemas().size();i++)
    {
        schemas.add( (PartitionSchema) transMeta.getPartitionSchemas().get(i).clone() );
    }
    previousSchemaIndex = -1;
    
    directoryChangeAllowed=true;
    changed=false;
}
 

import org.pentaho.di.partition.PartitionSchema; //导入依赖的package包/类
public synchronized long insertPartitionSchema(PartitionSchema partitionSchema) throws KettleException
{
    long id = getNextPartitionSchemaID();

    RowMetaAndData table = new RowMetaAndData();

    table.addValue(new ValueMeta(FIELD_PARTITION_SCHEMA_ID_PARTITION_SCHEMA, ValueMetaInterface.TYPE_INTEGER), new Long(id));
    table.addValue(new ValueMeta(FIELD_PARTITION_SCHEMA_NAME, ValueMetaInterface.TYPE_STRING), partitionSchema.getName());
    table.addValue(new ValueMeta(FIELD_PARTITION_SCHEMA_DYNAMIC_DEFINITION, ValueMetaInterface.TYPE_BOOLEAN), partitionSchema.isDynamicallyDefined());
    table.addValue(new ValueMeta(FIELD_PARTITION_SCHEMA_PARTITIONS_PER_SLAVE, ValueMetaInterface.TYPE_STRING), partitionSchema.getNumberOfPartitionsPerSlave());

    database.prepareInsert(table.getRowMeta(), TABLE_R_PARTITION_SCHEMA);
    database.setValuesInsert(table);
    database.insertRow();
    database.closeInsert();

    return id;
}
 

import org.pentaho.di.partition.PartitionSchema; //导入依赖的package包/类
public SlaveStepCopyPartitionDistribution(Node node) {
	this();
	
	int nrEntries = XMLHandler.countNodes(node, "entry");
	for (int i=0;i<nrEntries;i++) {
		Node entryNode = XMLHandler.getSubNodeByNr(node, "entry", i);
		String slaveServerName = XMLHandler.getTagValue(entryNode, "slavename");
		String partitionSchemaName = XMLHandler.getTagValue(entryNode, "partition_schema_name");
		int stepCopyNr = Const.toInt( XMLHandler.getTagValue(entryNode, "stepcopy"), -1);
		int partitionNr = Const.toInt( XMLHandler.getTagValue(entryNode, "partition"), -1);
		
		addPartition(slaveServerName, partitionSchemaName, stepCopyNr, partitionNr);
	}
	
	Node originalPartitionSchemasNode = XMLHandler.getSubNode(node, "original-partition-schemas");
	if (originalPartitionSchemasNode!=null) {
		originalPartitionSchemas = new ArrayList<PartitionSchema>();
		int nrSchemas = XMLHandler.countNodes(originalPartitionSchemasNode, PartitionSchema.XML_TAG);
		for (int i=0;i<nrSchemas;i++) {
			Node schemaNode = XMLHandler.getSubNodeByNr(originalPartitionSchemasNode, PartitionSchema.XML_TAG, i);
			PartitionSchema originalPartitionSchema = new PartitionSchema(schemaNode);
			originalPartitionSchemas.add(originalPartitionSchema);
		}
	}
}
 

import org.pentaho.di.partition.PartitionSchema; //导入依赖的package包/类
/**
    * Calculate the number of step copies in a step.<br>
    * If a step is not running clustered, it's simply returning getCopies().<br>
    * If a step is clustered and not doing any partitioning, it's simply returning getCopies().<br>
    * If a step is clustered and partitioned, we need to look in the partitioning map for the specified slave server.<br>
    * That is because the number of copies can vary over the slaves. (5 partitions over 3 slaves for example)
    * 
    * @param slaveServer the slave server
    * @param referenceStep the reference step
    * @return the number of step copies that we run.
    */
   private int determineNrOfStepCopies(SlaveServer slaveServer, StepMeta step) {
	if (!step.isClustered()) return step.getCopies();
	if (!step.isPartitioned()) return step.getCopies();
	if (slaveServer.isMaster()) return step.getCopies();
	
	// Partitioned and clustered...
	//
	StepPartitioningMeta stepPartitioningMeta = step.getStepPartitioningMeta();
	PartitionSchema partitionSchema = stepPartitioningMeta.getPartitionSchema();
	
	Map<PartitionSchema, List<String>> partitionMap = slaveServerPartitionsMap.get(slaveServer);
	List<String> partitionList = partitionMap.get(partitionSchema);
	
	return partitionList.size();
}
 

import org.pentaho.di.partition.PartitionSchema; //导入依赖的package包/类
/**
    * Create a copy of a step from the original transformation for use in the a slave transformation.
    * If the step is partitioned, the partitioning will be changed to "schemaName (slave)" 
    * 
    * @param stepMeta The step to copy / clone.
    * @return a copy of the specified step for use in a slave transformation.
    */
   private StepMeta addSlaveCopy(TransMeta transMeta, StepMeta stepMeta, SlaveServer slaveServer) {
	StepMeta copy = (StepMeta) stepMeta.clone();
	if (copy.isPartitioned()) {
		StepPartitioningMeta stepPartitioningMeta = copy.getStepPartitioningMeta();
		PartitionSchema partitionSchema = stepPartitioningMeta.getPartitionSchema();
		String slavePartitionSchemaName = createSlavePartitionSchemaName(partitionSchema.getName());
		PartitionSchema slaveSchema = transMeta.findPartitionSchema(slavePartitionSchemaName);
		if (slaveSchema!=null) {
			stepPartitioningMeta.setPartitionSchema(slaveSchema);
		}
		// Always just start a single copy on the slave server...
		// Otherwise the confusion w.r.t. to partitioning & re-partitioning would be complete.
		//
		copy.setCopies(1);
	}

	// Remove the clustering information on the slave transformation step
	// We don't need it anymore, it only confuses.
	//
	copy.setClusterSchema(null); 
	
	transMeta.addStep(copy);
	return copy;
}
 

import org.pentaho.di.partition.PartitionSchema; //导入依赖的package包/类
public boolean isUsingPartitionSchema(PartitionSchema partitionSchema)
{
    // Loop over all steps and see if the partition schema is used.
    for (int i=0;i<nrSteps();i++)
    {
        StepPartitioningMeta stepPartitioningMeta = getStep(i).getStepPartitioningMeta();
        if (stepPartitioningMeta!=null)
        {
            PartitionSchema check = stepPartitioningMeta.getPartitionSchema();
            if (check!=null && check.equals(partitionSchema))
            {
                return true;
            }
        }
    }
    return false;
}
 

import org.pentaho.di.partition.PartitionSchema; //导入依赖的package包/类
/**
 * Read the partitions in the repository and add them to this transformation if they are not yet present.
 * @param rep The repository to load from.
 * @param overWriteShared if an object with the same name exists, overwrite
 * @throws KettleException 
 */
public void readPartitionSchemas(Repository rep, boolean overWriteShared) throws KettleException
{
    try
    {
        long dbids[] = rep.getPartitionSchemaIDs();
        for (int i = 0; i < dbids.length; i++)
        {
            PartitionSchema partitionSchema = new PartitionSchema(rep, dbids[i]);
            PartitionSchema check = findPartitionSchema(partitionSchema.getName()); // Check if there already is one in the transformation
            if (check==null || overWriteShared) 
            {
                if (!Const.isEmpty(partitionSchema.getName()))
                {
                    addOrReplacePartitionSchema(partitionSchema);
                    if (!overWriteShared) partitionSchema.setChanged(false);
                }
            }
        }
    }
    catch (KettleException dbe)
    {
        throw new KettleException(Messages.getString("TransMeta.Log.UnableToReadPartitionSchemaFromRepository"), dbe); //$NON-NLS-1$
    }
}
 

import org.pentaho.di.partition.PartitionSchema; //导入依赖的package包/类
protected void applySchema()
{
    if (previousSchemaIndex>=0)
    {
        PartitionSchema partitionSchema = (PartitionSchema)schemas.get(previousSchemaIndex);
        partitionSchema.setName(wSchemaName.getText());
        java.util.List<String> ids = new ArrayList<String>();
        
        int nrNonEmptyPartitions = wPartitions.nrNonEmpty();
        for (int i=0;i<nrNonEmptyPartitions;i++)
        {
            ids.add( wPartitions.getNonEmpty(i).getText(1) );
        }
        partitionSchema.setPartitionIDs(ids);
    }
    previousSchemaIndex=wSchemaList.getSelectionIndex();
}
 

import org.pentaho.di.partition.PartitionSchema; //导入依赖的package包/类
protected void newSchema()
{
    String name = Messages.getString("TransDialog.NewPartitionSchema.Name", Integer.toString(schemas.size() + 1));
    EnterStringDialog askName = new EnterStringDialog(shell, name,
        Messages.getString("TransDialog.NewPartitionSchema.Title"),
        Messages.getString("TransDialog.NewPartitionSchema.Message"));
    name = askName.open();
    if (name!=null)
    {
        PartitionSchema schema = new PartitionSchema(name, new ArrayList<String>());
        schemas.add(schema);
        wSchemaList.add(name);
        previousSchemaIndex = schemas.size()-1; 
        wSchemaList.select( previousSchemaIndex );
        refreshPartitions();
    }
}
 

import org.pentaho.di.partition.PartitionSchema; //导入依赖的package包/类
public void shareObject(String id) {
	if ("database-inst-share".equals(id)) {
		final DatabaseMeta databaseMeta = (DatabaseMeta) selectionObject;
		shareObject(databaseMeta);
	}
	if ("step-inst-share".equals(id)) {
		final StepMeta stepMeta = (StepMeta) selectionObject;
		shareObject(stepMeta);
	}
	if ("partition-schema-inst-share".equals(id)) {
		final PartitionSchema partitionSchema = (PartitionSchema) selectionObject;
		shareObject(partitionSchema);
	}
	if ("cluster-schema-inst-share".equals(id)) {
		final ClusterSchema clusterSchema = (ClusterSchema) selectionObject;
		shareObject(clusterSchema);
	}
	if ("slave-server-inst-share".equals(id)) {
		final SlaveServer slaveServer = (SlaveServer) selectionObject;
		shareObject(slaveServer);
	}
}
 

import org.pentaho.di.partition.PartitionSchema; //导入依赖的package包/类
public PartitionSchema loadPartitionSchema(ObjectId id_partition_schema) throws KettleException
{
	PartitionSchema partitionSchema = new PartitionSchema();
       
	partitionSchema.setObjectId(id_partition_schema);
       
	RowMetaAndData row = getPartitionSchema(id_partition_schema);
       
	partitionSchema.setName( row.getString("NAME", null) );
       
	ObjectId[] pids = repository.getPartitionIDs(id_partition_schema);
	for (int i=0;i<pids.length;i++)
	{
		partitionSchema.getPartitionIDs().add( getPartition(pids[i]).getString("PARTITION_ID", null) );
	}
       
	partitionSchema.setDynamicallyDefined( row.getBoolean("DYNAMIC_DEFINITION", false) );
	partitionSchema.setNumberOfPartitionsPerSlave( row.getString("PARTITIONS_PER_SLAVE", null) );
	
	return partitionSchema;
}
 

import org.pentaho.di.partition.PartitionSchema; //导入依赖的package包/类
public synchronized ObjectId insertPartitionSchema(PartitionSchema partitionSchema) throws KettleException
{
  if(getPartitionSchemaID(partitionSchema.getName()) != null) {
    // This partition schema name is already in use. Throw an exception.
    throw new KettleObjectExistsException("Failed to create object in repository. Object [" + partitionSchema.getName() + "] already exists.");
  }
  
	ObjectId id = repository.connectionDelegate.getNextPartitionSchemaID();

    RowMetaAndData table = new RowMetaAndData();

    table.addValue(new ValueMeta(KettleDatabaseRepository.FIELD_PARTITION_SCHEMA_ID_PARTITION_SCHEMA, ValueMetaInterface.TYPE_INTEGER), id);
    table.addValue(new ValueMeta(KettleDatabaseRepository.FIELD_PARTITION_SCHEMA_NAME, ValueMetaInterface.TYPE_STRING), partitionSchema.getName());
    table.addValue(new ValueMeta(KettleDatabaseRepository.FIELD_PARTITION_SCHEMA_DYNAMIC_DEFINITION, ValueMetaInterface.TYPE_BOOLEAN), partitionSchema.isDynamicallyDefined());
    table.addValue(new ValueMeta(KettleDatabaseRepository.FIELD_PARTITION_SCHEMA_PARTITIONS_PER_SLAVE, ValueMetaInterface.TYPE_STRING), partitionSchema.getNumberOfPartitionsPerSlave());

    repository.connectionDelegate.getDatabase().prepareInsert(table.getRowMeta(), KettleDatabaseRepository.TABLE_R_PARTITION_SCHEMA);
    repository.connectionDelegate.getDatabase().setValuesInsert(table);
    repository.connectionDelegate.getDatabase().insertRow();
    repository.connectionDelegate.getDatabase().closeInsert();

    return id;
}
 

import org.pentaho.di.partition.PartitionSchema; //导入依赖的package包/类
/**
 * Read the partitions in the repository and add them to this transformation if they are not yet present.
 * @param TransMeta The transformation to load into.
 * @param overWriteShared if an object with the same name exists, overwrite
 * @throws KettleException 
 */
public void readPartitionSchemas(TransMeta transMeta, boolean overWriteShared) throws KettleException
{
    try
    {
 	   ObjectId dbids[] = repository.getPartitionSchemaIDs(false);
        for (int i = 0; i < dbids.length; i++)
        {
            PartitionSchema partitionSchema = repository.loadPartitionSchema(dbids[i], null);  // Load last version 
            PartitionSchema check = transMeta.findPartitionSchema(partitionSchema.getName()); // Check if there already is one in the transformation
            if (check==null || overWriteShared) 
            {
                if (!Const.isEmpty(partitionSchema.getName()))
                {
             	   transMeta.addOrReplacePartitionSchema(partitionSchema);
                    if (!overWriteShared) partitionSchema.setChanged(false);
                }
            }
        }
    }
    catch (KettleException dbe)
    {
        throw new KettleException(BaseMessages.getString(PKG, "TransMeta.Log.UnableToReadPartitionSchemaFromRepository"), dbe); //$NON-NLS-1$
    }
}
 

import org.pentaho.di.partition.PartitionSchema; //导入依赖的package包/类
/**
   * Calculate the number of step copies in a step.<br>
   * If a step is not running clustered, it's simply returning getCopies().<br>
   * If a step is clustered and not doing any partitioning, it's simply returning getCopies().<br>
   * If a step is clustered and partitioned, we need to look in the partitioning map for the specified slave server.<br>
   * That is because the number of copies can vary over the slaves. (5 partitions over 3 slaves for example)
   * 
   * @param slaveServer the slave server
   * @param referenceStep the reference step
   * @return the number of step copies that we run.
   */
  private int determineNrOfStepCopies(SlaveServer slaveServer, StepMeta step) {
  if (!step.isClustered()) return step.getCopies();
  if (!step.isPartitioned()) return step.getCopies();
  if (slaveServer.isMaster()) return step.getCopies();
  
  // Partitioned and clustered...
  //
  StepPartitioningMeta stepPartitioningMeta = step.getStepPartitioningMeta();
  PartitionSchema partitionSchema = stepPartitioningMeta.getPartitionSchema();
  
  Map<PartitionSchema, List<String>> partitionMap = slaveServerPartitionsMap.get(slaveServer);
  List<String> partitionList = partitionMap.get(partitionSchema);
  
  return partitionList.size();
}
 

import org.pentaho.di.partition.PartitionSchema; //导入依赖的package包/类
/**
   * Create a copy of a step from the original transformation for use in the a slave transformation.
   * If the step is partitioned, the partitioning will be changed to "schemaName (slave)" 
   * 
   * @param stepMeta The step to copy / clone.
   * @return a copy of the specified step for use in a slave transformation.
   */
  private StepMeta addSlaveCopy(TransMeta transMeta, StepMeta stepMeta, SlaveServer slaveServer) {
  StepMeta copy = (StepMeta) stepMeta.clone();
  if (copy.isPartitioned()) {
    StepPartitioningMeta stepPartitioningMeta = copy.getStepPartitioningMeta();
    PartitionSchema partitionSchema = stepPartitioningMeta.getPartitionSchema();
    String slavePartitionSchemaName = createSlavePartitionSchemaName(partitionSchema.getName());
    PartitionSchema slaveSchema = transMeta.findPartitionSchema(slavePartitionSchemaName);
    if (slaveSchema!=null) {
      stepPartitioningMeta.setPartitionSchema(slaveSchema);
    }
    // Always just start a single copy on the slave server...
    // Otherwise the confusion w.r.t. to partitioning & re-partitioning would be complete.
    //
    copy.setCopies(1);
  }

  // Remove the clustering information on the slave transformation step
  // We don't need it anymore, it only confuses.
  //
  copy.setClusterSchema(null); 
  
  transMeta.addStep(copy);
  return copy;
}
 

import org.pentaho.di.partition.PartitionSchema; //导入依赖的package包/类
public void editPartitionSchema(String partitionSchemaName)
{
    try
    {
        ObjectId id = rep.getPartitionSchemaID(partitionSchemaName);
        PartitionSchema partitionSchema = rep.loadPartitionSchema(id, null);  // Load the last version

        PartitionSchemaDialog dd = new PartitionSchemaDialog(shell, partitionSchema, rep.readDatabases(), variableSpace);
        if (dd.open())
        {
            rep.insertLogEntry("Updating partition schema '"+partitionSchema.getName()+"'");
            rep.save(partitionSchema, Const.VERSION_COMMENT_EDIT_VERSION, null);
            if(!partitionSchemaName.equalsIgnoreCase(partitionSchema.getName())) refreshTree();
        }
    }
    catch(KettleException e)
    {
        new ErrorDialog(shell, BaseMessages.getString(PKG, "RepositoryExplorerDialog.PartitionSchema.Edit.UnexpectedError.Title"), 
                BaseMessages.getString(PKG, "RepositoryExplorerDialog.PartitionSchema.Edit.UnexpectedError.Message")+partitionSchemaName+"]", e); //$NON-NLS-1$ //$NON-NLS-2$ //$NON-NLS-3$
    }
}
 

import org.pentaho.di.partition.PartitionSchema; //导入依赖的package包/类
public void refreshPartitions() {
  if (repository != null) {
    final List<UIPartition> tmpList = new ArrayList<UIPartition>();
    Runnable r = new Runnable() {
      public void run() {
        try {
          ObjectId[] partitionIdList = repository.getPartitionSchemaIDs(false);

          for (ObjectId partitionId : partitionIdList) {
            PartitionSchema partition = repository.loadPartitionSchema(partitionId, null);
            // Add the partition schema to the list
            tmpList.add(new UIPartition(partition));
          }
        } catch (KettleException e) {
          // convert to runtime exception so it bubbles up through the UI
          throw new RuntimeException(e);
        }
      
      }
    };
    doWithBusyIndicator(r);
    partitionList.setChildren(tmpList);
  }
}
 

import org.pentaho.di.partition.PartitionSchema; //导入依赖的package包/类
public void shareObject(String id) {
  if ("database-inst-share".equals(id)) {
    final DatabaseMeta databaseMeta = (DatabaseMeta) selectionObject;
    if (databaseMeta.isShared()) {
      unShareObject(databaseMeta);
    } else {
      shareObject(databaseMeta);
    }
  }
  if ("step-inst-share".equals(id)) {
    final StepMeta stepMeta = (StepMeta) selectionObject;
    shareObject(stepMeta);
  }
  if ("partition-schema-inst-share".equals(id)) {
    final PartitionSchema partitionSchema = (PartitionSchema) selectionObject;
    shareObject(partitionSchema);
  }
  if ("cluster-schema-inst-share".equals(id)) {
    final ClusterSchema clusterSchema = (ClusterSchema) selectionObject;
    shareObject(clusterSchema);
  }
  if ("slave-server-inst-share".equals(id)) {
    final SlaveServer slaveServer = (SlaveServer) selectionObject;
    shareObject(slaveServer);
  }
}
 

import org.pentaho.di.partition.PartitionSchema; //导入依赖的package包/类
/**
 * This creates a new partitioning schema, edits it and adds it to the
 * transformation metadata
 * 
 */
public void newPartitioningSchema(TransMeta transMeta) {
  PartitionSchema partitionSchema = new PartitionSchema();

  PartitionSchemaDialog dialog = new PartitionSchemaDialog(shell, partitionSchema, transMeta.getDatabases(),
      transMeta);
  if (dialog.open()) {
    transMeta.getPartitionSchemas().add(partitionSchema);
    
    if (rep!=null) {
      try {
        if (!rep.getSecurityProvider().isReadOnly()) {
          rep.save(partitionSchema, Const.VERSION_COMMENT_INITIAL_VERSION, null);
        } else {
          throw new KettleException(BaseMessages.getString(PKG, "Spoon.Dialog.Exception.ReadOnlyRepositoryUser"));
        }
      } catch (KettleException e) {
        new ErrorDialog(getShell(), BaseMessages.getString(PKG, "Spoon.Dialog.ErrorSavingPartition.Title"), BaseMessages.getString(PKG, "Spoon.Dialog.ErrorSavingPartition.Message", partitionSchema.getName()), e);
      }
    }
    
    refreshTree();
  }
}
 

import org.pentaho.di.partition.PartitionSchema; //导入依赖的package包/类
public void delPartitionSchema( TransMeta transMeta, PartitionSchema partitionSchema ) {
  try {
    int idx = transMeta.getPartitionSchemas().indexOf( partitionSchema );
    transMeta.getPartitionSchemas().remove( idx );

    if ( spoon.rep != null && partitionSchema.getObjectId() != null ) {
      // remove the partition schema from the repository too...
      spoon.rep.deletePartitionSchema( partitionSchema.getObjectId() );
      if ( sharedObjectSyncUtil != null ) {
        sharedObjectSyncUtil.deletePartitionSchema( partitionSchema );
      }
    }
    spoon.refreshTree();
  } catch ( KettleException e ) {
    new ErrorDialog(
      spoon.getShell(), BaseMessages.getString( PKG, "Spoon.Dialog.ErrorDeletingClusterSchema.Title" ), BaseMessages
        .getString( PKG, "Spoon.Dialog.ErrorDeletingClusterSchema.Message" ), e );
  }
}